2798: Full integration test with real API server and Keep server succeeds.

Expanded return values a bit to include the URL that a block was actually
downloaded from on GET, and the number of replicas written on PUT.

diff --git a/sdk/go/src/arvados.org/keepclient/keepclient.go b/sdk/go/src/arvados.org/keepclient/keepclient.go
index 50d1bd761210b95cb7dec63e023e52749dd2dff3..aeb805b7a0db9c6d1f3b2af97599612c61acf4a7 100644 (file)
--- a/sdk/go/src/arvados.org/keepclient/keepclient.go
+++ b/sdk/go/src/arvados.org/keepclient/keepclient.go
@@ -32,9 +32,10 @@ type KeepDisk struct {
 
 func MakeKeepClient() (kc *KeepClient, err error) {
        kc = &KeepClient{
-               ApiServer:   os.Getenv("ARVADOS_API_HOST"),
-               ApiToken:    os.Getenv("ARVADOS_API_TOKEN"),
-               ApiInsecure: (os.Getenv("ARVADOS_API_HOST_INSECURE") != "")}
+               ApiServer:     os.Getenv("ARVADOS_API_HOST"),
+               ApiToken:      os.Getenv("ARVADOS_API_TOKEN"),
+               ApiInsecure:   (os.Getenv("ARVADOS_API_HOST_INSECURE") != ""),
+               Want_replicas: 2}
 
        tr := &http.Transport{
                TLSClientConfig: &tls.Config{InsecureSkipVerify: kc.ApiInsecure},
@@ -432,7 +433,7 @@ func (this KeepClient) putReplicas(
        hash string,
        requests chan ReadRequest,
        reader_status chan error,
-       expectedLength int64) error {
+       expectedLength int64) (replicas int, err error) {
 
        // Calculate the ordering for uploading to servers
        sv := this.ShuffledServiceRoots(hash)
@@ -448,17 +449,17 @@ func (this KeepClient) putReplicas(
        defer close(upload_status)
 
        // Desired number of replicas
-       want_replicas := this.Want_replicas
+       remaining_replicas := this.Want_replicas
 
-       for want_replicas > 0 {
-               for active < want_replicas {
+       for remaining_replicas > 0 {
+               for active < remaining_replicas {
                        // Start some upload requests
                        if next_server < len(sv) {
                                go this.uploadToKeepServer(sv[next_server], hash, MakeBufferReader(requests), upload_status, expectedLength)
                                next_server += 1
                                active += 1
                        } else {
-                               return InsufficientReplicasError
+                               return (this.Want_replicas - remaining_replicas), InsufficientReplicasError
                        }
                }
 
@@ -469,34 +470,34 @@ func (this KeepClient) putReplicas(
                                // good news!
                        } else {
                                // bad news
-                               return status
+                               return (this.Want_replicas - remaining_replicas), status
                        }
                case status := <-upload_status:
                        if status.StatusCode == 200 {
                                // good news!
-                               want_replicas -= 1
+                               remaining_replicas -= 1
                        } else {
                                // writing to keep server failed for some reason
                                log.Printf("Keep server put to %v failed with '%v'",
                                        status.Url, status.Err)
                        }
                        active -= 1
-                       log.Printf("Upload status %v %v %v", status.StatusCode, want_replicas, active)
+                       log.Printf("Upload status %v %v %v", status.StatusCode, remaining_replicas, active)
                }
        }
 
-       return nil
+       return (this.Want_replicas - remaining_replicas), nil
 }
 
 var OversizeBlockError = errors.New("Block too big")
 
-func (this KeepClient) PutHR(hash string, r io.Reader, expectedLength int64) error {
+func (this KeepClient) PutHR(hash string, r io.Reader, expectedLength int64) (replicas int, err error) {
 
        // Buffer for reads from 'r'
        var buffer []byte
        if expectedLength > 0 {
                if expectedLength > 64*1024*1024 {
-                       return OversizeBlockError
+                       return 0, OversizeBlockError
                }
                buffer = make([]byte, expectedLength)
        } else {
@@ -517,7 +518,7 @@ func (this KeepClient) PutHR(hash string, r io.Reader, expectedLength int64) err
        return this.putReplicas(hash, requests, reader_status, expectedLength)
 }
 
-func (this KeepClient) PutHB(hash string, buffer []byte) error {
+func (this KeepClient) PutHB(hash string, buffer []byte) (replicas int, err error) {
        // Read requests on Transfer() buffer
        requests := make(chan ReadRequest)
        defer close(requests)
@@ -528,13 +529,15 @@ func (this KeepClient) PutHB(hash string, buffer []byte) error {
        return this.putReplicas(hash, requests, nil, int64(len(buffer)))
 }
 
-func (this KeepClient) PutB(buffer []byte) error {
-       return this.PutHB(fmt.Sprintf("%x", md5.Sum(buffer)), buffer)
+func (this KeepClient) PutB(buffer []byte) (hash string, replicas int, err error) {
+       hash = fmt.Sprintf("%x", md5.Sum(buffer))
+       replicas, err = this.PutHB(hash, buffer)
+       return hash, replicas, err
 }
 
-func (this KeepClient) PutR(r io.Reader) error {
+func (this KeepClient) PutR(r io.Reader) (hash string, replicas int, err error) {
        if buffer, err := ioutil.ReadAll(r); err != nil {
-               return err
+               return "", 0, err
        } else {
                return this.PutB(buffer)
        }

diff --git a/sdk/go/src/arvados.org/keepclient/keepclient_test.go b/sdk/go/src/arvados.org/keepclient/keepclient_test.go
index 2c7b87728ee81b8266c2a11df47c0f149a35db54..00a2063cb6b46094dc42a3456382c2c7c17ae35c 100644 (file)
--- a/sdk/go/src/arvados.org/keepclient/keepclient_test.go
+++ b/sdk/go/src/arvados.org/keepclient/keepclient_test.go
@@ -48,7 +48,7 @@ func (s *ServerRequiredSuite) TearDownSuite(c *C) {
        exec.Command("python", "run_test_server.py", "stop").Run()
 }
 
-func (s *ServerRequiredSuite) TestInit(c *C) {
+func (s *ServerRequiredSuite) TestMakeKeepClient(c *C) {
        os.Setenv("ARVADOS_API_HOST", "localhost:3001")
        os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
        os.Setenv("ARVADOS_API_HOST_INSECURE", "")
@@ -712,11 +712,13 @@ func (s *StandaloneSuite) TestPutWithFail(c *C) {
 
        shuff := kc.ShuffledServiceRoots(fmt.Sprintf("%x", md5.Sum([]byte("foo"))))
 
-       err := kc.PutB([]byte("foo"))
+       phash, replicas, err := kc.PutB([]byte("foo"))
 
        <-fh.handled
 
        c.Check(err, Equals, nil)
+       c.Check(phash, Equals, hash)
+       c.Check(replicas, Equals, 2)
        c.Check(<-st.handled, Equals, shuff[1])
        c.Check(<-st.handled, Equals, shuff[2])
 }
@@ -758,9 +760,10 @@ func (s *StandaloneSuite) TestPutWithTooManyFail(c *C) {
 
        shuff := kc.ShuffledServiceRoots(fmt.Sprintf("%x", md5.Sum([]byte("foo"))))
 
-       err := kc.PutB([]byte("foo"))
+       _, replicas, err := kc.PutB([]byte("foo"))
 
        c.Check(err, Equals, InsufficientReplicasError)
+       c.Check(replicas, Equals, 1)
        c.Check(<-st.handled, Equals, shuff[1])
 }
 
@@ -865,3 +868,26 @@ func (s *StandaloneSuite) TestGetWithFailures(c *C) {
        c.Check(err2, Equals, nil)
        c.Check(content, DeepEquals, []byte("foo"))
 }
+
+func (s *ServerRequiredSuite) TestPutAndGet(c *C) {
+       os.Setenv("ARVADOS_API_HOST", "localhost:3001")
+       os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
+       os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
+
+       kc, err := MakeKeepClient()
+       c.Assert(err, Equals, nil)
+
+       hash, replicas, err := kc.PutB([]byte("foo"))
+       c.Check(hash, Equals, fmt.Sprintf("%x", md5.Sum([]byte("foo"))))
+       c.Check(replicas, Equals, 2)
+       c.Check(err, Equals, nil)
+
+       r, n, url2, err := kc.Get(hash)
+       c.Check(err, Equals, nil)
+       c.Check(n, Equals, int64(3))
+       c.Check(url2, Equals, fmt.Sprintf("http://localhost:25108/%s", hash))
+
+       content, err2 := ioutil.ReadAll(r)
+       c.Check(err2, Equals, nil)
+       c.Check(content, DeepEquals, []byte("foo"))
+}