[ARVADOS] updated: 7d5d57a522489209e6b3cecfef94bab0aae4a7f5
git at public.curoverse.com
git at public.curoverse.com
Thu Dec 10 16:16:23 EST 2015
Summary of changes:
sdk/go/arvadostest/fixtures.go | 9 +
sdk/go/keepclient/discover.go | 4 +-
sdk/go/keepclient/keepclient.go | 3 +-
sdk/go/keepclient/keepclient_test.go | 37 +--
sdk/go/keepclient/support.go | 15 +-
services/keepstore/azure_blob_volume.go | 2 +-
services/keepstore/azure_blob_volume_test.go | 8 +-
services/keepstore/bufferpool_test.go | 6 -
services/keepstore/collision_test.go | 6 -
services/keepstore/gocheck_test.go | 10 +
.../keepstore/handlers_with_generic_volume_test.go | 15 +-
services/keepstore/keepstore_test.go | 8 +-
services/keepstore/pull_worker_test.go | 11 +-
services/keepstore/s3_volume.go | 312 +++++++++++++++++++++
services/keepstore/s3_volume_test.go | 133 +++++++++
services/keepstore/volume_generic_test.go | 76 ++---
services/keepstore/volume_unix_test.go | 12 +-
tools/keep-exercise/keep-exercise.go | 25 ++
18 files changed, 568 insertions(+), 124 deletions(-)
create mode 100644 services/keepstore/gocheck_test.go
create mode 100644 services/keepstore/s3_volume.go
create mode 100644 services/keepstore/s3_volume_test.go
via 7d5d57a522489209e6b3cecfef94bab0aae4a7f5 (commit)
via f14632e1c7ddc7a8c7a7c0d1535acb003600a1d5 (commit)
via ac89a2193e0ca105b986190cb6de7112f4b1ebdd (commit)
via 980564a2c5d0b28e1b3b84da102992c45f194f1b (commit)
from 809a3785297f5460602888c2ddacec6fe0976589 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit 7d5d57a522489209e6b3cecfef94bab0aae4a7f5
Merge: 809a378 f14632e
Author: Tom Clegg <tom at curoverse.com>
Date: Thu Dec 10 16:12:20 2015 -0500
Merge branch '7393-s3-volume' closes #7393
commit f14632e1c7ddc7a8c7a7c0d1535acb003600a1d5
Author: Tom Clegg <tom at curoverse.com>
Date: Wed Dec 9 16:00:34 2015 -0500
7393: Add -uuid and -url options, fix memory sharing in -vary-request.
diff --git a/tools/keep-exercise/keep-exercise.go b/tools/keep-exercise/keep-exercise.go
index a94c01e..9dc8f94 100644
--- a/tools/keep-exercise/keep-exercise.go
+++ b/tools/keep-exercise/keep-exercise.go
@@ -36,6 +36,8 @@ var (
VaryThread = flag.Bool("vary-thread", false, "use -wthreads different data blocks")
Replicas = flag.Int("replicas", 1, "replication level for writing")
StatsInterval = flag.Duration("stats-interval", time.Second, "time interval between IO stats reports, or 0 to disable")
+ ServiceURL = flag.String("url", "", "specify scheme://host of a single keep service to exercise (instead of using all advertised services like normal clients)")
+ ServiceUUID = flag.String("uuid", "", "specify UUID of a single advertised keep service to exercise")
)
func main() {
@@ -52,6 +54,8 @@ func main() {
kc.Want_replicas = *Replicas
kc.Client.Timeout = 10 * time.Minute
+ overrideServices(kc)
+
nextBuf := make(chan []byte, *WriteThreads)
nextLocator := make(chan string, *ReadThreads+*WriteThreads)
@@ -109,6 +113,7 @@ func makeBufs(nextBuf chan []byte, threadID int) {
}
for {
if *VaryRequest {
+ buf = make([]byte, *BlockSize)
if _, err := io.ReadFull(rand.Reader, buf); err != nil {
log.Fatal(err)
}
@@ -155,3 +160,23 @@ func doReads(kc *keepclient.KeepClient, nextLocator chan string) {
bytesInChan <- uint64(n)
}
}
+
+func overrideServices(kc *keepclient.KeepClient) {
+ roots := make(map[string]string)
+ if *ServiceURL != "" {
+ roots["zzzzz-bi6l4-000000000000000"] = *ServiceURL
+ } else if *ServiceUUID != "" {
+ for uuid, url := range kc.GatewayRoots() {
+ if uuid == *ServiceUUID {
+ roots[uuid] = url
+ break
+ }
+ }
+ if len(roots) == 0 {
+ log.Fatalf("Service %q was not in list advertised by API %+q", *ServiceUUID, kc.GatewayRoots())
+ }
+ } else {
+ return
+ }
+ kc.SetServiceRoots(roots, roots, roots)
+}
commit ac89a2193e0ca105b986190cb6de7112f4b1ebdd
Author: Tom Clegg <tom at curoverse.com>
Date: Wed Dec 9 15:59:54 2015 -0500
7393: Quiet excessive debug printfs.
diff --git a/sdk/go/keepclient/discover.go b/sdk/go/keepclient/discover.go
index 099c56f..f039c21 100644
--- a/sdk/go/keepclient/discover.go
+++ b/sdk/go/keepclient/discover.go
@@ -57,14 +57,14 @@ func (kc *KeepClient) RefreshServices(interval, errInterval time.Duration) {
timer.Reset(interval)
if err := kc.DiscoverKeepServers(); err != nil {
- log.Println("Error retrieving services list: %v (retrying in %v)", err, errInterval)
+ log.Printf("WARNING: Error retrieving services list: %v (retrying in %v)", err, errInterval)
timer.Reset(errInterval)
continue
}
newRoots := []map[string]string{kc.LocalRoots(), kc.GatewayRoots()}
if !reflect.DeepEqual(previousRoots, newRoots) {
- log.Printf("Updated services list: locals %v gateways %v", newRoots[0], newRoots[1])
+ DebugPrintf("DEBUG: Updated services list: locals %v gateways %v", newRoots[0], newRoots[1])
previousRoots = newRoots
}
diff --git a/sdk/go/keepclient/keepclient.go b/sdk/go/keepclient/keepclient.go
index f15a6b2..26aa717 100644
--- a/sdk/go/keepclient/keepclient.go
+++ b/sdk/go/keepclient/keepclient.go
@@ -11,7 +11,6 @@ import (
"git.curoverse.com/arvados.git/sdk/go/streamer"
"io"
"io/ioutil"
- "log"
"net/http"
"regexp"
"strconv"
@@ -233,7 +232,7 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i
}
serversToTry = retryList
}
- log.Printf("DEBUG: %s %s failed: %v", method, locator, errs)
+ DebugPrintf("DEBUG: %s %s failed: %v", method, locator, errs)
var err error
if count404 == numServers {
diff --git a/sdk/go/keepclient/keepclient_test.go b/sdk/go/keepclient/keepclient_test.go
index 87b9b1d..4ba1d7c 100644
--- a/sdk/go/keepclient/keepclient_test.go
+++ b/sdk/go/keepclient/keepclient_test.go
@@ -155,13 +155,9 @@ func (s *StandaloneSuite) TestUploadToStubKeepServer(c *C) {
status := <-upload_status
c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, ""})
})
-
- log.Printf("TestUploadToStubKeepServer done")
}
func (s *StandaloneSuite) TestUploadToStubKeepServerBufferReader(c *C) {
- log.Printf("TestUploadToStubKeepServerBufferReader")
-
st := StubPutHandler{
c,
"acbd18db4cc2f85cedef654fccc4a4d8",
@@ -188,8 +184,6 @@ func (s *StandaloneSuite) TestUploadToStubKeepServerBufferReader(c *C) {
status := <-upload_status
c.Check(status, DeepEquals, uploadStatus{nil, fmt.Sprintf("%s/%s", url, st.expectPath), 200, 1, ""})
})
-
- log.Printf("TestUploadToStubKeepServerBufferReader done")
}
type FailHandler struct {
@@ -227,8 +221,6 @@ func (fh Error404Handler) ServeHTTP(resp http.ResponseWriter, req *http.Request)
}
func (s *StandaloneSuite) TestFailedUploadToStubKeepServer(c *C) {
- log.Printf("TestFailedUploadToStubKeepServer")
-
st := FailHandler{
make(chan string)}
@@ -249,7 +241,6 @@ func (s *StandaloneSuite) TestFailedUploadToStubKeepServer(c *C) {
c.Check(status.url, Equals, fmt.Sprintf("%s/%s", url, hash))
c.Check(status.statusCode, Equals, 500)
})
- log.Printf("TestFailedUploadToStubKeepServer done")
}
type KeepServer struct {
@@ -268,8 +259,6 @@ func RunSomeFakeKeepServers(st http.Handler, n int) (ks []KeepServer) {
}
func (s *StandaloneSuite) TestPutB(c *C) {
- log.Printf("TestPutB")
-
hash := Md5String("foo")
st := StubPutHandler{
@@ -308,13 +297,9 @@ func (s *StandaloneSuite) TestPutB(c *C) {
(s1 == shuff[1] && s2 == shuff[0]),
Equals,
true)
-
- log.Printf("TestPutB done")
}
func (s *StandaloneSuite) TestPutHR(c *C) {
- log.Printf("TestPutHR")
-
hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
st := StubPutHandler{
@@ -352,7 +337,6 @@ func (s *StandaloneSuite) TestPutHR(c *C) {
kc.PutHR(hash, reader, 3)
shuff := NewRootSorter(kc.LocalRoots(), hash).GetSortedRoots()
- log.Print(shuff)
s1 := <-st.handled
s2 := <-st.handled
@@ -361,13 +345,9 @@ func (s *StandaloneSuite) TestPutHR(c *C) {
(s1 == shuff[1] && s2 == shuff[0]),
Equals,
true)
-
- log.Printf("TestPutHR done")
}
func (s *StandaloneSuite) TestPutWithFail(c *C) {
- log.Printf("TestPutWithFail")
-
hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
st := StubPutHandler{
@@ -406,6 +386,7 @@ func (s *StandaloneSuite) TestPutWithFail(c *C) {
shuff := NewRootSorter(
kc.LocalRoots(), Md5String("foo")).GetSortedRoots()
+ c.Logf("%+v", shuff)
phash, replicas, err := kc.PutB([]byte("foo"))
@@ -425,8 +406,6 @@ func (s *StandaloneSuite) TestPutWithFail(c *C) {
}
func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
- log.Printf("TestPutWithTooManyFail")
-
hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
st := StubPutHandler{
@@ -469,8 +448,6 @@ func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
c.Check(err, Equals, InsufficientReplicasError)
c.Check(replicas, Equals, 1)
c.Check(<-st.handled, Equals, ks1[0].url)
-
- log.Printf("TestPutWithTooManyFail done")
}
type StubGetHandler struct {
@@ -490,8 +467,6 @@ func (sgh StubGetHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request)
}
func (s *StandaloneSuite) TestGet(c *C) {
- log.Printf("TestGet")
-
hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
st := StubGetHandler{
@@ -518,8 +493,6 @@ func (s *StandaloneSuite) TestGet(c *C) {
content, err2 := ioutil.ReadAll(r)
c.Check(err2, Equals, nil)
c.Check(content, DeepEquals, []byte("foo"))
-
- log.Printf("TestGet done")
}
func (s *StandaloneSuite) TestGet404(c *C) {
@@ -887,8 +860,6 @@ func (this StubProxyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reque
}
func (s *StandaloneSuite) TestPutProxy(c *C) {
- log.Printf("TestPutProxy")
-
st := StubProxyHandler{make(chan string, 1)}
arv, err := arvadosclient.MakeArvadosClient()
@@ -914,13 +885,9 @@ func (s *StandaloneSuite) TestPutProxy(c *C) {
c.Check(err, Equals, nil)
c.Check(replicas, Equals, 2)
-
- log.Printf("TestPutProxy done")
}
func (s *StandaloneSuite) TestPutProxyInsufficientReplicas(c *C) {
- log.Printf("TestPutProxy")
-
st := StubProxyHandler{make(chan string, 1)}
arv, err := arvadosclient.MakeArvadosClient()
@@ -945,8 +912,6 @@ func (s *StandaloneSuite) TestPutProxyInsufficientReplicas(c *C) {
c.Check(err, Equals, InsufficientReplicasError)
c.Check(replicas, Equals, 2)
-
- log.Printf("TestPutProxy done")
}
func (s *StandaloneSuite) TestMakeLocator(c *C) {
diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go
index b904b09..b12f512 100644
--- a/sdk/go/keepclient/support.go
+++ b/sdk/go/keepclient/support.go
@@ -7,7 +7,6 @@ import (
"git.curoverse.com/arvados.git/sdk/go/streamer"
"io"
"io/ioutil"
- "log"
"math/rand"
"net"
"net/http"
@@ -101,7 +100,7 @@ func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.Rea
var err error
var url = fmt.Sprintf("%s/%s", host, hash)
if req, err = http.NewRequest("PUT", url, nil); err != nil {
- log.Printf("[%08x] Error creating request PUT %v error: %v", requestID, url, err.Error())
+ DebugPrintf("DEBUG: [%08x] Error creating request PUT %v error: %v", requestID, url, err.Error())
upload_status <- uploadStatus{err, url, 0, 0, ""}
body.Close()
return
@@ -126,7 +125,7 @@ func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.Rea
var resp *http.Response
if resp, err = this.Client.Do(req); err != nil {
- log.Printf("[%08x] Upload failed %v error: %v", requestID, url, err.Error())
+ DebugPrintf("DEBUG: [%08x] Upload failed %v error: %v", requestID, url, err.Error())
upload_status <- uploadStatus{err, url, 0, 0, ""}
return
}
@@ -142,13 +141,13 @@ func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.Rea
respbody, err2 := ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
response := strings.TrimSpace(string(respbody))
if err2 != nil && err2 != io.EOF {
- log.Printf("[%08x] Upload %v error: %v response: %v", requestID, url, err2.Error(), response)
+ DebugPrintf("DEBUG: [%08x] Upload %v error: %v response: %v", requestID, url, err2.Error(), response)
upload_status <- uploadStatus{err2, url, resp.StatusCode, rep, response}
} else if resp.StatusCode == http.StatusOK {
- log.Printf("[%08x] Upload %v success", requestID, url)
+ DebugPrintf("DEBUG: [%08x] Upload %v success", requestID, url)
upload_status <- uploadStatus{nil, url, resp.StatusCode, rep, response}
} else {
- log.Printf("[%08x] Upload %v error: %v response: %v", requestID, url, resp.StatusCode, response)
+ DebugPrintf("DEBUG: [%08x] Upload %v error: %v response: %v", requestID, url, resp.StatusCode, response)
upload_status <- uploadStatus{errors.New(resp.Status), url, resp.StatusCode, rep, response}
}
}
@@ -205,7 +204,7 @@ func (this *KeepClient) putReplicas(
for active*replicasPerThread < remaining_replicas {
// Start some upload requests
if next_server < len(sv) {
- log.Printf("[%08x] Begin upload %s to %s", requestID, hash, sv[next_server])
+ DebugPrintf("DEBUG: [%08x] Begin upload %s to %s", requestID, hash, sv[next_server])
go this.uploadToKeepServer(sv[next_server], hash, tr.MakeStreamReader(), upload_status, expectedLength, requestID)
next_server += 1
active += 1
@@ -217,7 +216,7 @@ func (this *KeepClient) putReplicas(
}
}
}
- log.Printf("[%08x] Replicas remaining to write: %v active uploads: %v",
+ DebugPrintf("DEBUG: [%08x] Replicas remaining to write: %v active uploads: %v",
requestID, remaining_replicas, active)
// Now wait for something to happen.
commit 980564a2c5d0b28e1b3b84da102992c45f194f1b
Author: Tom Clegg <tom at curoverse.com>
Date: Wed Dec 9 12:43:50 2015 -0500
7393: Add S3 volume type.
diff --git a/sdk/go/arvadostest/fixtures.go b/sdk/go/arvadostest/fixtures.go
index 3256ec2..47b75b3 100644
--- a/sdk/go/arvadostest/fixtures.go
+++ b/sdk/go/arvadostest/fixtures.go
@@ -24,3 +24,12 @@ const PathologicalManifest = ". acbd18db4cc2f85cedef654fccc4a4d8+3 37b51d194a751
`./foo\040b\141r acbd18db4cc2f85cedef654fccc4a4d8+3 0:3:b\141z\040w\141z` + "\n" +
"./foo acbd18db4cc2f85cedef654fccc4a4d8+3 0:0:zero 0:3:foo\n" +
". acbd18db4cc2f85cedef654fccc4a4d8+3 0:0:foo/zero 0:3:foo/foo\n"
+
+// An MD5 collision.
+var (
+ MD5CollisionData = [][]byte{
+ []byte("\x0e0eaU\x9a\xa7\x87\xd0\x0b\xc6\xf7\x0b\xbd\xfe4\x04\xcf\x03e\x9epO\x854\xc0\x0f\xfbe\x9cL\x87@\xcc\x94/\xeb-\xa1\x15\xa3\xf4\x15\\\xbb\x86\x07Is\x86em}\x1f4\xa4 Y\xd7\x8fZ\x8d\xd1\xef"),
+ []byte("\x0e0eaU\x9a\xa7\x87\xd0\x0b\xc6\xf7\x0b\xbd\xfe4\x04\xcf\x03e\x9etO\x854\xc0\x0f\xfbe\x9cL\x87@\xcc\x94/\xeb-\xa1\x15\xa3\xf4\x15\xdc\xbb\x86\x07Is\x86em}\x1f4\xa4 Y\xd7\x8fZ\x8d\xd1\xef"),
+ }
+ MD5CollisionMD5 = "cee9a457e790cf20d4bdaa6d69f01e41"
+)
diff --git a/services/keepstore/azure_blob_volume.go b/services/keepstore/azure_blob_volume.go
index e9fda2a..0f98e6e 100644
--- a/services/keepstore/azure_blob_volume.go
+++ b/services/keepstore/azure_blob_volume.go
@@ -189,7 +189,7 @@ func (v *AzureBlobVolume) Compare(loc string, expect []byte) error {
return compareReaderWithBuf(rdr, expect, loc[:32])
}
-// Put sotres a Keep block as a block blob in the container.
+// Put stores a Keep block as a block blob in the container.
func (v *AzureBlobVolume) Put(loc string, block []byte) error {
if v.readonly {
return MethodDisabledError
diff --git a/services/keepstore/azure_blob_volume_test.go b/services/keepstore/azure_blob_volume_test.go
index a240c23..b8bf5cb 100644
--- a/services/keepstore/azure_blob_volume_test.go
+++ b/services/keepstore/azure_blob_volume_test.go
@@ -292,10 +292,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)
@@ -341,7 +341,7 @@ 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)
})
}
@@ -355,7 +355,7 @@ 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)
})
}
diff --git a/services/keepstore/bufferpool_test.go b/services/keepstore/bufferpool_test.go
index 8726a19..7b51b64 100644
--- a/services/keepstore/bufferpool_test.go
+++ b/services/keepstore/bufferpool_test.go
@@ -2,15 +2,9 @@ package main
import (
. "gopkg.in/check.v1"
- "testing"
"time"
)
-// Gocheck boilerplate
-func TestBufferPool(t *testing.T) {
- TestingT(t)
-}
-
var _ = Suite(&BufferPoolSuite{})
type BufferPoolSuite struct{}
diff --git a/services/keepstore/collision_test.go b/services/keepstore/collision_test.go
index 379dadd..d9b7e61 100644
--- a/services/keepstore/collision_test.go
+++ b/services/keepstore/collision_test.go
@@ -2,17 +2,11 @@ package main
import (
"bytes"
- "testing"
"testing/iotest"
check "gopkg.in/check.v1"
)
-// Gocheck boilerplate
-func Test(t *testing.T) {
- check.TestingT(t)
-}
-
var _ = check.Suite(&CollisionSuite{})
type CollisionSuite struct{}
diff --git a/services/keepstore/gocheck_test.go b/services/keepstore/gocheck_test.go
new file mode 100644
index 0000000..133ed6e
--- /dev/null
+++ b/services/keepstore/gocheck_test.go
@@ -0,0 +1,10 @@
+package main
+
+import (
+ "gopkg.in/check.v1"
+ "testing"
+)
+
+func TestGocheck(t *testing.T) {
+ check.TestingT(t)
+}
diff --git a/services/keepstore/handlers_with_generic_volume_test.go b/services/keepstore/handlers_with_generic_volume_test.go
index 9f31f5f..c5349d3 100644
--- a/services/keepstore/handlers_with_generic_volume_test.go
+++ b/services/keepstore/handlers_with_generic_volume_test.go
@@ -2,19 +2,18 @@ package main
import (
"bytes"
- "testing"
)
// A TestableVolumeManagerFactory creates a volume manager with at least two TestableVolume instances.
// The factory function, and the TestableVolume instances it returns, can use "t" to write
// logs, fail the current test, etc.
-type TestableVolumeManagerFactory func(t *testing.T) (*RRVolumeManager, []TestableVolume)
+type TestableVolumeManagerFactory func(t TB) (*RRVolumeManager, []TestableVolume)
// DoHandlersWithGenericVolumeTests runs a set of handler tests with a
// Volume Manager comprised of TestableVolume instances.
// It calls factory to create a volume manager with TestableVolume
// instances for each test case, to avoid leaking state between tests.
-func DoHandlersWithGenericVolumeTests(t *testing.T, factory TestableVolumeManagerFactory) {
+func DoHandlersWithGenericVolumeTests(t TB, factory TestableVolumeManagerFactory) {
testGetBlock(t, factory, TestHash, TestBlock)
testGetBlock(t, factory, EmptyHash, EmptyBlock)
testPutRawBadDataGetBlock(t, factory, TestHash, TestBlock, []byte("baddata"))
@@ -26,7 +25,7 @@ func DoHandlersWithGenericVolumeTests(t *testing.T, factory TestableVolumeManage
}
// Setup RRVolumeManager with TestableVolumes
-func setupHandlersWithGenericVolumeTest(t *testing.T, factory TestableVolumeManagerFactory) []TestableVolume {
+func setupHandlersWithGenericVolumeTest(t TB, factory TestableVolumeManagerFactory) []TestableVolume {
vm, testableVolumes := factory(t)
KeepVM = vm
@@ -39,7 +38,7 @@ func setupHandlersWithGenericVolumeTest(t *testing.T, factory TestableVolumeMana
}
// Put a block using PutRaw in just one volume and Get it using GetBlock
-func testGetBlock(t *testing.T, factory TestableVolumeManagerFactory, testHash string, testBlock []byte) {
+func testGetBlock(t TB, factory TestableVolumeManagerFactory, testHash string, testBlock []byte) {
testableVolumes := setupHandlersWithGenericVolumeTest(t, factory)
// Put testBlock in one volume
@@ -56,7 +55,7 @@ func testGetBlock(t *testing.T, factory TestableVolumeManagerFactory, testHash s
}
// Put a bad block using PutRaw and get it.
-func testPutRawBadDataGetBlock(t *testing.T, factory TestableVolumeManagerFactory,
+func testPutRawBadDataGetBlock(t TB, factory TestableVolumeManagerFactory,
testHash string, testBlock []byte, badData []byte) {
testableVolumes := setupHandlersWithGenericVolumeTest(t, factory)
@@ -72,7 +71,7 @@ func testPutRawBadDataGetBlock(t *testing.T, factory TestableVolumeManagerFactor
}
// Invoke PutBlock twice to ensure CompareAndTouch path is tested.
-func testPutBlock(t *testing.T, factory TestableVolumeManagerFactory, testHash string, testBlock []byte) {
+func testPutBlock(t TB, factory TestableVolumeManagerFactory, testHash string, testBlock []byte) {
setupHandlersWithGenericVolumeTest(t, factory)
// PutBlock
@@ -95,7 +94,7 @@ func testPutBlock(t *testing.T, factory TestableVolumeManagerFactory, testHash s
}
// Put a bad block using PutRaw, overwrite it using PutBlock and get it.
-func testPutBlockCorrupt(t *testing.T, factory TestableVolumeManagerFactory,
+func testPutBlockCorrupt(t TB, factory TestableVolumeManagerFactory,
testHash string, testBlock []byte, badData []byte) {
testableVolumes := setupHandlersWithGenericVolumeTest(t, factory)
diff --git a/services/keepstore/keepstore_test.go b/services/keepstore/keepstore_test.go
index 8a004b7..2a1c3d2 100644
--- a/services/keepstore/keepstore_test.go
+++ b/services/keepstore/keepstore_test.go
@@ -10,6 +10,8 @@ import (
"sort"
"strings"
"testing"
+
+ "git.curoverse.com/arvados.git/sdk/go/arvadostest"
)
var TestBlock = []byte("The quick brown fox jumps over the lazy dog.")
@@ -229,9 +231,9 @@ func TestPutBlockCollision(t *testing.T) {
defer teardown()
// These blocks both hash to the MD5 digest cee9a457e790cf20d4bdaa6d69f01e41.
- var b1 = []byte("\x0e0eaU\x9a\xa7\x87\xd0\x0b\xc6\xf7\x0b\xbd\xfe4\x04\xcf\x03e\x9epO\x854\xc0\x0f\xfbe\x9cL\x87@\xcc\x94/\xeb-\xa1\x15\xa3\xf4\x15\\\xbb\x86\x07Is\x86em}\x1f4\xa4 Y\xd7\x8fZ\x8d\xd1\xef")
- var b2 = []byte("\x0e0eaU\x9a\xa7\x87\xd0\x0b\xc6\xf7\x0b\xbd\xfe4\x04\xcf\x03e\x9etO\x854\xc0\x0f\xfbe\x9cL\x87@\xcc\x94/\xeb-\xa1\x15\xa3\xf4\x15\xdc\xbb\x86\x07Is\x86em}\x1f4\xa4 Y\xd7\x8fZ\x8d\xd1\xef")
- var locator = "cee9a457e790cf20d4bdaa6d69f01e41"
+ b1 := arvadostest.MD5CollisionData[0]
+ b2 := arvadostest.MD5CollisionData[1]
+ locator := arvadostest.MD5CollisionMD5
// Prepare two test Keep volumes.
KeepVM = MakeTestVolumeManager(2)
diff --git a/services/keepstore/pull_worker_test.go b/services/keepstore/pull_worker_test.go
index c6a4195..5076b85 100644
--- a/services/keepstore/pull_worker_test.go
+++ b/services/keepstore/pull_worker_test.go
@@ -8,20 +8,13 @@ import (
. "gopkg.in/check.v1"
"io"
"net/http"
- "testing"
"time"
)
-type PullWorkerTestSuite struct{}
-
-// Gocheck boilerplate
-func TestPullWorker(t *testing.T) {
- TestingT(t)
-}
-
-// Gocheck boilerplate
var _ = Suite(&PullWorkerTestSuite{})
+type PullWorkerTestSuite struct{}
+
var testPullLists map[string]string
var readContent string
var readError error
diff --git a/services/keepstore/s3_volume.go b/services/keepstore/s3_volume.go
new file mode 100644
index 0000000..572ee46
--- /dev/null
+++ b/services/keepstore/s3_volume.go
@@ -0,0 +1,312 @@
+package main
+
+import (
+ "encoding/base64"
+ "encoding/hex"
+ "flag"
+ "fmt"
+ "io"
+ "log"
+ "net/http"
+ "os"
+ "regexp"
+ "time"
+
+ "github.com/AdRoll/goamz/aws"
+ "github.com/AdRoll/goamz/s3"
+)
+
+var (
+ ErrS3DeleteNotAvailable = fmt.Errorf("delete without -s3-unsafe-delete is not implemented")
+
+ s3AccessKeyFile string
+ s3SecretKeyFile string
+ s3RegionName string
+ s3Endpoint string
+ s3Replication int
+ s3UnsafeDelete bool
+
+ s3ACL = s3.Private
+)
+
+const (
+ maxClockSkew = 600 * time.Second
+ nearlyRFC1123 = "Mon, 2 Jan 2006 15:04:05 GMT"
+)
+
+type s3VolumeAdder struct {
+ *volumeSet
+}
+
+func (s *s3VolumeAdder) Set(bucketName string) error {
+ if bucketName == "" {
+ return fmt.Errorf("no container name given")
+ }
+ if s3AccessKeyFile == "" || s3SecretKeyFile == "" {
+ return fmt.Errorf("-s3-access-key-file and -s3-secret-key-file arguments must given before -s3-bucket-volume")
+ }
+ region, ok := aws.Regions[s3RegionName]
+ if s3Endpoint == "" {
+ if !ok {
+ return fmt.Errorf("unrecognized region %+q; try specifying -s3-endpoint instead", s3RegionName)
+ }
+ } else {
+ if ok {
+ return fmt.Errorf("refusing to use AWS region name %+q with endpoint %+q; "+
+ "specify empty endpoint (\"-s3-endpoint=\") or use a different region name", s3RegionName, s3Endpoint)
+ }
+ region = aws.Region{
+ Name: s3RegionName,
+ S3Endpoint: s3Endpoint,
+ }
+ }
+ var err error
+ var auth aws.Auth
+ auth.AccessKey, err = readKeyFromFile(s3AccessKeyFile)
+ if err != nil {
+ return err
+ }
+ auth.SecretKey, err = readKeyFromFile(s3SecretKeyFile)
+ if err != nil {
+ return err
+ }
+ if flagSerializeIO {
+ log.Print("Notice: -serialize is not supported by s3-bucket volumes.")
+ }
+ v := NewS3Volume(auth, region, bucketName, flagReadonly, s3Replication)
+ if err := v.Check(); err != nil {
+ return err
+ }
+ *s.volumeSet = append(*s.volumeSet, v)
+ return nil
+}
+
+func s3regions() (okList []string) {
+ for r, _ := range aws.Regions {
+ okList = append(okList, r)
+ }
+ return
+}
+
+func init() {
+ flag.Var(&s3VolumeAdder{&volumes},
+ "s3-bucket-volume",
+ "Use the given bucket as a storage volume. Can be given multiple times.")
+ flag.StringVar(
+ &s3RegionName,
+ "s3-region",
+ "",
+ fmt.Sprintf("AWS region used for subsequent -s3-bucket-volume arguments. Allowed values are %+q.", s3regions()))
+ flag.StringVar(
+ &s3Endpoint,
+ "s3-endpoint",
+ "",
+ "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\".")
+ flag.StringVar(
+ &s3AccessKeyFile,
+ "s3-access-key-file",
+ "",
+ "File containing the access key used for subsequent -s3-bucket-volume arguments.")
+ flag.StringVar(
+ &s3SecretKeyFile,
+ "s3-secret-key-file",
+ "",
+ "File containing the secret key used for subsequent -s3-bucket-volume arguments.")
+ flag.IntVar(
+ &s3Replication,
+ "s3-replication",
+ 2,
+ "Replication level reported to clients for subsequent -s3-bucket-volume arguments.")
+ flag.BoolVar(
+ &s3UnsafeDelete,
+ "s3-unsafe-delete",
+ false,
+ "EXPERIMENTAL. Enable deletion (garbage collection), even though there are known race conditions that can cause data loss.")
+}
+
+type S3Volume struct {
+ *s3.Bucket
+ readonly bool
+ replication int
+ indexPageSize int
+}
+
+// NewS3Volume returns a new S3Volume using the given auth, region,
+// and bucket name. The replication argument specifies the replication
+// level to report when writing data.
+func NewS3Volume(auth aws.Auth, region aws.Region, bucket string, readonly bool, replication int) *S3Volume {
+ return &S3Volume{
+ Bucket: &s3.Bucket{
+ S3: s3.New(auth, region),
+ Name: bucket,
+ },
+ readonly: readonly,
+ replication: replication,
+ indexPageSize: 1000,
+ }
+}
+
+func (v *S3Volume) Check() error {
+ return nil
+}
+
+func (v *S3Volume) Get(loc string) ([]byte, error) {
+ rdr, err := v.Bucket.GetReader(loc)
+ if err != nil {
+ return nil, v.translateError(err)
+ }
+ defer rdr.Close()
+ buf := bufs.Get(BlockSize)
+ n, err := io.ReadFull(rdr, buf)
+ switch err {
+ case nil, io.EOF, io.ErrUnexpectedEOF:
+ return buf[:n], nil
+ default:
+ bufs.Put(buf)
+ return nil, v.translateError(err)
+ }
+}
+
+func (v *S3Volume) Compare(loc string, expect []byte) error {
+ rdr, err := v.Bucket.GetReader(loc)
+ if err != nil {
+ return v.translateError(err)
+ }
+ defer rdr.Close()
+ return v.translateError(compareReaderWithBuf(rdr, expect, loc[:32]))
+}
+
+func (v *S3Volume) Put(loc string, block []byte) error {
+ if v.readonly {
+ return MethodDisabledError
+ }
+ var opts s3.Options
+ if len(block) > 0 {
+ md5, err := hex.DecodeString(loc)
+ if err != nil {
+ return err
+ }
+ opts.ContentMD5 = base64.StdEncoding.EncodeToString(md5)
+ }
+ return v.translateError(
+ v.Bucket.Put(
+ loc, block, "application/octet-stream", s3ACL, opts))
+}
+
+func (v *S3Volume) Touch(loc string) error {
+ if v.readonly {
+ return MethodDisabledError
+ }
+ result, err := v.Bucket.PutCopy(loc, s3ACL, s3.CopyOptions{
+ ContentType: "application/octet-stream",
+ MetadataDirective: "REPLACE",
+ }, v.Bucket.Name+"/"+loc)
+ if err != nil {
+ return v.translateError(err)
+ }
+ t, err := time.Parse(time.RFC3339, result.LastModified)
+ if err != nil {
+ return err
+ }
+ if time.Since(t) > maxClockSkew {
+ return fmt.Errorf("PutCopy returned old LastModified %s => %s (%s ago)", result.LastModified, t, time.Since(t))
+ }
+ return nil
+}
+
+func (v *S3Volume) Mtime(loc string) (time.Time, error) {
+ resp, err := v.Bucket.Head(loc, nil)
+ if err != nil {
+ return zeroTime, v.translateError(err)
+ }
+ hdr := resp.Header.Get("Last-Modified")
+ t, err := time.Parse(time.RFC1123, hdr)
+ if err != nil && hdr != "" {
+ // AWS example is "Sun, 1 Jan 2006 12:00:00 GMT",
+ // which isn't quite "Sun, 01 Jan 2006 12:00:00 GMT"
+ // as required by HTTP spec. If it's not a valid HTTP
+ // header value, it's probably AWS (or s3test) giving
+ // us a nearly-RFC1123 timestamp.
+ t, err = time.Parse(nearlyRFC1123, hdr)
+ }
+ return t, err
+}
+
+func (v *S3Volume) IndexTo(prefix string, writer io.Writer) error {
+ nextMarker := ""
+ for {
+ listResp, err := v.Bucket.List(prefix, "", nextMarker, v.indexPageSize)
+ if err != nil {
+ return err
+ }
+ for _, key := range listResp.Contents {
+ t, err := time.Parse(time.RFC3339, key.LastModified)
+ if err != nil {
+ return err
+ }
+ if !v.isKeepBlock(key.Key) {
+ continue
+ }
+ fmt.Fprintf(writer, "%s+%d %d\n", key.Key, key.Size, t.Unix())
+ }
+ if !listResp.IsTruncated {
+ break
+ }
+ nextMarker = listResp.NextMarker
+ }
+ return nil
+}
+
+func (v *S3Volume) Delete(loc string) error {
+ if v.readonly {
+ return MethodDisabledError
+ }
+ if t, err := v.Mtime(loc); err != nil {
+ return err
+ } else if time.Since(t) < blobSignatureTTL {
+ return nil
+ }
+ if !s3UnsafeDelete {
+ return ErrS3DeleteNotAvailable
+ }
+ return v.Bucket.Del(loc)
+}
+
+func (v *S3Volume) Status() *VolumeStatus {
+ return &VolumeStatus{
+ DeviceNum: 1,
+ BytesFree: BlockSize * 1000,
+ BytesUsed: 1,
+ }
+}
+
+func (v *S3Volume) String() string {
+ return fmt.Sprintf("s3-bucket:%+q", v.Bucket.Name)
+}
+
+func (v *S3Volume) Writable() bool {
+ return !v.readonly
+}
+func (v *S3Volume) Replication() int {
+ return v.replication
+}
+
+var s3KeepBlockRegexp = regexp.MustCompile(`^[0-9a-f]{32}$`)
+
+func (v *S3Volume) isKeepBlock(s string) bool {
+ return s3KeepBlockRegexp.MatchString(s)
+}
+
+func (v *S3Volume) translateError(err error) error {
+ switch err := err.(type) {
+ case *s3.Error:
+ if err.StatusCode == http.StatusNotFound && err.Code == "NoSuchKey" {
+ return os.ErrNotExist
+ }
+ // Other 404 errors like NoSuchVersion and
+ // NoSuchBucket are different problems which should
+ // get called out downstream, so we don't convert them
+ // to os.ErrNotExist.
+ }
+ return err
+}
diff --git a/services/keepstore/s3_volume_test.go b/services/keepstore/s3_volume_test.go
new file mode 100644
index 0000000..0c2cd49
--- /dev/null
+++ b/services/keepstore/s3_volume_test.go
@@ -0,0 +1,133 @@
+package main
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "strings"
+ "time"
+
+ "github.com/AdRoll/goamz/aws"
+ "github.com/AdRoll/goamz/s3"
+ "github.com/AdRoll/goamz/s3/s3test"
+ check "gopkg.in/check.v1"
+)
+
+type TestableS3Volume struct {
+ *S3Volume
+ server *s3test.Server
+ c *check.C
+ serverClock *fakeClock
+}
+
+const (
+ TestBucketName = "testbucket"
+)
+
+type fakeClock struct {
+ now *time.Time
+}
+
+func (c *fakeClock) Now() time.Time {
+ if c.now == nil {
+ return time.Now()
+ }
+ return *c.now
+}
+
+func init() {
+ // Deleting isn't safe from races, but if it's turned on
+ // anyway we do expect it to pass the generic volume tests.
+ s3UnsafeDelete = true
+}
+
+func NewTestableS3Volume(c *check.C, readonly bool, replication int) *TestableS3Volume {
+ clock := &fakeClock{}
+ srv, err := s3test.NewServer(&s3test.Config{Clock: clock})
+ c.Assert(err, check.IsNil)
+ auth := aws.Auth{}
+ region := aws.Region{
+ Name: "test-region-1",
+ S3Endpoint: srv.URL(),
+ S3LocationConstraint: true,
+ }
+ bucket := &s3.Bucket{
+ S3: s3.New(auth, region),
+ Name: TestBucketName,
+ }
+ err = bucket.PutBucket(s3.ACL("private"))
+ c.Assert(err, check.IsNil)
+
+ return &TestableS3Volume{
+ S3Volume: NewS3Volume(auth, region, TestBucketName, readonly, replication),
+ server: srv,
+ serverClock: clock,
+ }
+}
+
+var _ = check.Suite(&StubbedS3Suite{})
+
+type StubbedS3Suite struct {
+ volumes []*TestableS3Volume
+}
+
+func (s *StubbedS3Suite) TestGeneric(c *check.C) {
+ DoGenericVolumeTests(c, func(t TB) TestableVolume {
+ return NewTestableS3Volume(c, false, 2)
+ })
+}
+
+func (s *StubbedS3Suite) TestGenericReadOnly(c *check.C) {
+ DoGenericVolumeTests(c, func(t TB) TestableVolume {
+ return NewTestableS3Volume(c, true, 2)
+ })
+}
+
+func (s *StubbedS3Suite) TestIndex(c *check.C) {
+ v := NewTestableS3Volume(c, false, 2)
+ v.indexPageSize = 3
+ for i := 0; i < 256; i++ {
+ v.PutRaw(fmt.Sprintf("%02x%030x", i, i), []byte{102, 111, 111})
+ }
+ for _, spec := range []struct {
+ prefix string
+ expectMatch int
+ }{
+ {"", 256},
+ {"c", 16},
+ {"bc", 1},
+ {"abc", 0},
+ } {
+ buf := new(bytes.Buffer)
+ err := v.IndexTo(spec.prefix, buf)
+ c.Check(err, check.IsNil)
+
+ idx := bytes.SplitAfter(buf.Bytes(), []byte{10})
+ c.Check(len(idx), check.Equals, spec.expectMatch+1)
+ c.Check(len(idx[len(idx)-1]), check.Equals, 0)
+ }
+}
+
+// PutRaw skips the ContentMD5 test
+func (v *TestableS3Volume) PutRaw(loc string, block []byte) {
+ err := v.Bucket.Put(loc, block, "application/octet-stream", s3ACL, s3.Options{})
+ if err != nil {
+ log.Printf("PutRaw: %+v", err)
+ }
+}
+
+// TouchWithDate turns back the clock while doing a Touch(). We assume
+// there are no other operations happening on the same s3test server
+// while we do this.
+func (v *TestableS3Volume) TouchWithDate(locator string, lastPut time.Time) {
+ v.serverClock.now = &lastPut
+ err := v.Touch(locator)
+ if err != nil && !strings.Contains(err.Error(), "PutCopy returned old LastModified") {
+ log.Printf("Touch: %+v", err)
+ }
+ v.serverClock.now = nil
+}
+
+func (v *TestableS3Volume) Teardown() {
+ v.server.Quit()
+}
diff --git a/services/keepstore/volume_generic_test.go b/services/keepstore/volume_generic_test.go
index 61088f1..fae4a9e 100644
--- a/services/keepstore/volume_generic_test.go
+++ b/services/keepstore/volume_generic_test.go
@@ -8,19 +8,32 @@ import (
"regexp"
"sort"
"strings"
- "testing"
"time"
+
+ "git.curoverse.com/arvados.git/sdk/go/arvadostest"
)
+type TB interface {
+ Error(args ...interface{})
+ Errorf(format string, args ...interface{})
+ Fail()
+ FailNow()
+ Failed() bool
+ Fatal(args ...interface{})
+ Fatalf(format string, args ...interface{})
+ Log(args ...interface{})
+ Logf(format string, args ...interface{})
+}
+
// A TestableVolumeFactory returns a new TestableVolume. The factory
// function, and the TestableVolume it returns, can use "t" to write
// logs, fail the current test, etc.
-type TestableVolumeFactory func(t *testing.T) TestableVolume
+type TestableVolumeFactory func(t TB) TestableVolume
// DoGenericVolumeTests runs a set of tests that every TestableVolume
// is expected to pass. It calls factory to create a new TestableVolume
// for each test case, to avoid leaking state between tests.
-func DoGenericVolumeTests(t *testing.T, factory TestableVolumeFactory) {
+func DoGenericVolumeTests(t TB, factory TestableVolumeFactory) {
testGet(t, factory)
testGetNoSuchBlock(t, factory)
@@ -36,10 +49,10 @@ func DoGenericVolumeTests(t *testing.T, factory TestableVolumeFactory) {
testPutBlockWithSameContent(t, factory, TestHash, TestBlock)
testPutBlockWithSameContent(t, factory, EmptyHash, EmptyBlock)
- testPutBlockWithDifferentContent(t, factory, TestHash, TestBlock, TestBlock2)
- testPutBlockWithDifferentContent(t, factory, TestHash, EmptyBlock, TestBlock)
- testPutBlockWithDifferentContent(t, factory, TestHash, TestBlock, EmptyBlock)
- testPutBlockWithDifferentContent(t, factory, EmptyHash, EmptyBlock, TestBlock)
+ testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, arvadostest.MD5CollisionData[0], arvadostest.MD5CollisionData[1])
+ testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, EmptyBlock, arvadostest.MD5CollisionData[0])
+ testPutBlockWithDifferentContent(t, factory, arvadostest.MD5CollisionMD5, arvadostest.MD5CollisionData[0], EmptyBlock)
+ testPutBlockWithDifferentContent(t, factory, EmptyHash, EmptyBlock, arvadostest.MD5CollisionData[0])
testPutMultipleBlocks(t, factory)
testPutAndTouch(t, factory)
@@ -67,7 +80,7 @@ func DoGenericVolumeTests(t *testing.T, factory TestableVolumeFactory) {
// Put a test block, get it and verify content
// Test should pass for both writable and read-only volumes
-func testGet(t *testing.T, factory TestableVolumeFactory) {
+func testGet(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -87,7 +100,7 @@ func testGet(t *testing.T, factory TestableVolumeFactory) {
// Invoke get on a block that does not exist in volume; should result in error
// Test should pass for both writable and read-only volumes
-func testGetNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
+func testGetNoSuchBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -99,7 +112,7 @@ func testGetNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
// Compare() should return os.ErrNotExist if the block does not exist.
// Otherwise, writing new data causes CompareAndTouch() to generate
// error logs even though everything is working fine.
-func testCompareNonexistent(t *testing.T, factory TestableVolumeFactory) {
+func testCompareNonexistent(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -111,7 +124,7 @@ func testCompareNonexistent(t *testing.T, factory TestableVolumeFactory) {
// Put a test block and compare the locator with same content
// Test should pass for both writable and read-only volumes
-func testCompareSameContent(t *testing.T, factory TestableVolumeFactory, testHash string, testData []byte) {
+func testCompareSameContent(t TB, factory TestableVolumeFactory, testHash string, testData []byte) {
v := factory(t)
defer v.Teardown()
@@ -129,7 +142,7 @@ func testCompareSameContent(t *testing.T, factory TestableVolumeFactory, testHas
// testHash = md5(testDataA).
//
// Test should pass for both writable and read-only volumes
-func testCompareWithCollision(t *testing.T, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) {
+func testCompareWithCollision(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) {
v := factory(t)
defer v.Teardown()
@@ -146,7 +159,7 @@ func testCompareWithCollision(t *testing.T, factory TestableVolumeFactory, testH
// corrupted. Requires testHash = md5(testDataA) != md5(testDataB).
//
// Test should pass for both writable and read-only volumes
-func testCompareWithCorruptStoredData(t *testing.T, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) {
+func testCompareWithCorruptStoredData(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) {
v := factory(t)
defer v.Teardown()
@@ -160,7 +173,7 @@ func testCompareWithCorruptStoredData(t *testing.T, factory TestableVolumeFactor
// Put a block and put again with same content
// Test is intended for only writable volumes
-func testPutBlockWithSameContent(t *testing.T, factory TestableVolumeFactory, testHash string, testData []byte) {
+func testPutBlockWithSameContent(t TB, factory TestableVolumeFactory, testHash string, testData []byte) {
v := factory(t)
defer v.Teardown()
@@ -181,7 +194,7 @@ func testPutBlockWithSameContent(t *testing.T, factory TestableVolumeFactory, te
// Put a block and put again with different content
// Test is intended for only writable volumes
-func testPutBlockWithDifferentContent(t *testing.T, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) {
+func testPutBlockWithDifferentContent(t TB, factory TestableVolumeFactory, testHash string, testDataA, testDataB []byte) {
v := factory(t)
defer v.Teardown()
@@ -189,10 +202,7 @@ func testPutBlockWithDifferentContent(t *testing.T, factory TestableVolumeFactor
return
}
- err := v.Put(testHash, testDataA)
- if err != nil {
- t.Errorf("Got err putting block %q: %q, expected nil", testDataA, err)
- }
+ v.PutRaw(testHash, testDataA)
putErr := v.Put(testHash, testDataB)
buf, getErr := v.Get(testHash)
@@ -217,7 +227,7 @@ func testPutBlockWithDifferentContent(t *testing.T, factory TestableVolumeFactor
// Put and get multiple blocks
// Test is intended for only writable volumes
-func testPutMultipleBlocks(t *testing.T, factory TestableVolumeFactory) {
+func testPutMultipleBlocks(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -275,7 +285,7 @@ func testPutMultipleBlocks(t *testing.T, factory TestableVolumeFactory) {
// Test that when applying PUT to a block that already exists,
// the block's modification time is updated.
// Test is intended for only writable volumes
-func testPutAndTouch(t *testing.T, factory TestableVolumeFactory) {
+func testPutAndTouch(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -317,7 +327,7 @@ func testPutAndTouch(t *testing.T, factory TestableVolumeFactory) {
// Touching a non-existing block should result in error.
// Test should pass for both writable and read-only volumes
-func testTouchNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
+func testTouchNoSuchBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -328,7 +338,7 @@ func testTouchNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
// Invoking Mtime on a non-existing block should result in error.
// Test should pass for both writable and read-only volumes
-func testMtimeNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
+func testMtimeNoSuchBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -342,7 +352,7 @@ func testMtimeNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
// * with a prefix
// * with no such prefix
// Test should pass for both writable and read-only volumes
-func testIndexTo(t *testing.T, factory TestableVolumeFactory) {
+func testIndexTo(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -399,7 +409,7 @@ func testIndexTo(t *testing.T, factory TestableVolumeFactory) {
// Calling Delete() for a block immediately after writing it (not old enough)
// should neither delete the data nor return an error.
// Test is intended for only writable volumes
-func testDeleteNewBlock(t *testing.T, factory TestableVolumeFactory) {
+func testDeleteNewBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
blobSignatureTTL = 300 * time.Second
@@ -427,7 +437,7 @@ func testDeleteNewBlock(t *testing.T, factory TestableVolumeFactory) {
// Calling Delete() for a block with a timestamp older than
// blobSignatureTTL seconds in the past should delete the data.
// Test is intended for only writable volumes
-func testDeleteOldBlock(t *testing.T, factory TestableVolumeFactory) {
+func testDeleteOldBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
blobSignatureTTL = 300 * time.Second
@@ -449,7 +459,7 @@ func testDeleteOldBlock(t *testing.T, factory TestableVolumeFactory) {
// Calling Delete() for a block that does not exist should result in error.
// Test should pass for both writable and read-only volumes
-func testDeleteNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
+func testDeleteNoSuchBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -460,7 +470,7 @@ func testDeleteNoSuchBlock(t *testing.T, factory TestableVolumeFactory) {
// Invoke Status and verify that VolumeStatus is returned
// Test should pass for both writable and read-only volumes
-func testStatus(t *testing.T, factory TestableVolumeFactory) {
+func testStatus(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -481,7 +491,7 @@ func testStatus(t *testing.T, factory TestableVolumeFactory) {
// Invoke String for the volume; expect non-empty result
// Test should pass for both writable and read-only volumes
-func testString(t *testing.T, factory TestableVolumeFactory) {
+func testString(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -492,7 +502,7 @@ func testString(t *testing.T, factory TestableVolumeFactory) {
// Putting, updating, touching, and deleting blocks from a read-only volume result in error.
// Test is intended for only read-only volumes
-func testUpdateReadOnly(t *testing.T, factory TestableVolumeFactory) {
+func testUpdateReadOnly(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -539,7 +549,7 @@ func testUpdateReadOnly(t *testing.T, factory TestableVolumeFactory) {
// Launch concurrent Gets
// Test should pass for both writable and read-only volumes
-func testGetConcurrent(t *testing.T, factory TestableVolumeFactory) {
+func testGetConcurrent(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -592,7 +602,7 @@ func testGetConcurrent(t *testing.T, factory TestableVolumeFactory) {
// Launch concurrent Puts
// Test is intended for only writable volumes
-func testPutConcurrent(t *testing.T, factory TestableVolumeFactory) {
+func testPutConcurrent(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
@@ -660,7 +670,7 @@ func testPutConcurrent(t *testing.T, factory TestableVolumeFactory) {
}
// Write and read back a full size block
-func testPutFullBlock(t *testing.T, factory TestableVolumeFactory) {
+func testPutFullBlock(t TB, factory TestableVolumeFactory) {
v := factory(t)
defer v.Teardown()
diff --git a/services/keepstore/volume_unix_test.go b/services/keepstore/volume_unix_test.go
index 924637f..b216810 100644
--- a/services/keepstore/volume_unix_test.go
+++ b/services/keepstore/volume_unix_test.go
@@ -16,10 +16,10 @@ import (
type TestableUnixVolume struct {
UnixVolume
- t *testing.T
+ t TB
}
-func NewTestableUnixVolume(t *testing.T, serialize bool, readonly bool) *TestableUnixVolume {
+func NewTestableUnixVolume(t TB, serialize bool, readonly bool) *TestableUnixVolume {
d, err := ioutil.TempDir("", "volume_test")
if err != nil {
t.Fatal(err)
@@ -66,28 +66,28 @@ func (v *TestableUnixVolume) Teardown() {
// serialize = false; readonly = false
func TestUnixVolumeWithGenericTests(t *testing.T) {
- DoGenericVolumeTests(t, func(t *testing.T) TestableVolume {
+ DoGenericVolumeTests(t, func(t TB) TestableVolume {
return NewTestableUnixVolume(t, false, false)
})
}
// serialize = false; readonly = true
func TestUnixVolumeWithGenericTestsReadOnly(t *testing.T) {
- DoGenericVolumeTests(t, func(t *testing.T) TestableVolume {
+ DoGenericVolumeTests(t, func(t TB) TestableVolume {
return NewTestableUnixVolume(t, false, true)
})
}
// serialize = true; readonly = false
func TestUnixVolumeWithGenericTestsSerialized(t *testing.T) {
- DoGenericVolumeTests(t, func(t *testing.T) TestableVolume {
+ DoGenericVolumeTests(t, func(t TB) TestableVolume {
return NewTestableUnixVolume(t, true, false)
})
}
// serialize = false; readonly = false
func TestUnixVolumeHandlersWithGenericVolumeTests(t *testing.T) {
- DoHandlersWithGenericVolumeTests(t, func(t *testing.T) (*RRVolumeManager, []TestableVolume) {
+ DoHandlersWithGenericVolumeTests(t, func(t TB) (*RRVolumeManager, []TestableVolume) {
vols := make([]Volume, 2)
testableUnixVols := make([]TestableVolume, 2)
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list