9701: Merge branch '9463-change-arvput-use-collection-class' into 9701-collection...

[arvados.git] / services / keepstore / pull_worker.go

diff --git a/services/keepstore/pull_worker.go b/services/keepstore/pull_worker.go
index d85458a325a1c44e2e53d177da6bd12f8adbe07b..d53d1060e743e07d9d2bfba6b90c67376a1006ab 100644 (file)
--- a/services/keepstore/pull_worker.go
+++ b/services/keepstore/pull_worker.go
@@ -2,7 +2,6 @@ package main
 
 import (
        "crypto/rand"
-       "errors"
        "fmt"
        "git.curoverse.com/arvados.git/sdk/go/keepclient"
        "io"
@@ -11,19 +10,19 @@ import (
        "time"
 )
 
-/*
-       Keepstore initiates pull worker channel goroutine.
-       The channel will process pull list.
-               For each (next) pull request:
-                       For each locator listed, execute Pull on the server(s) listed
-                       Skip the rest of the servers if no errors
-               Repeat
-*/
+// RunPullWorker is used by Keepstore to initiate pull worker channel goroutine.
+//     The channel will process pull list.
+//             For each (next) pull request:
+//                     For each locator listed, execute Pull on the server(s) listed
+//                     Skip the rest of the servers if no errors
+//             Repeat
+//
 func RunPullWorker(pullq *WorkQueue, keepClient *keepclient.KeepClient) {
        nextItem := pullq.NextItem
        for item := range nextItem {
                pullRequest := item.(PullRequest)
-               err := PullItemAndProcess(item.(PullRequest), GenerateRandomApiToken(), keepClient)
+               err := PullItemAndProcess(item.(PullRequest), GenerateRandomAPIToken(), keepClient)
+               pullq.DoneItem <- struct{}{}
                if err == nil {
                        log.Printf("Pull %s success", pullRequest)
                } else {
@@ -32,45 +31,45 @@ func RunPullWorker(pullq *WorkQueue, keepClient *keepclient.KeepClient) {
        }
 }
 
-/*
-       For each Pull request:
-               Generate a random API token.
-               Generate a permission signature using this token, timestamp ~60 seconds in the future, and desired block hash.
-               Using this token & signature, retrieve the given block.
-               Write to storage
-*/
+// PullItemAndProcess pulls items from PullQueue and processes them.
+//     For each Pull request:
+//             Generate a random API token.
+//             Generate a permission signature using this token, timestamp ~60 seconds in the future, and desired block hash.
+//             Using this token & signature, retrieve the given block.
+//             Write to storage
+//
 func PullItemAndProcess(pullRequest PullRequest, token string, keepClient *keepclient.KeepClient) (err error) {
        keepClient.Arvados.ApiToken = token
 
-       service_roots := make(map[string]string)
+       serviceRoots := make(map[string]string)
        for _, addr := range pullRequest.Servers {
-               service_roots[addr] = addr
+               serviceRoots[addr] = addr
        }
-       keepClient.SetServiceRoots(service_roots, nil)
+       keepClient.SetServiceRoots(serviceRoots, nil, nil)
 
        // Generate signature with a random token
-       expires_at := time.Now().Add(60 * time.Second)
-       signedLocator := SignLocator(pullRequest.Locator, token, expires_at)
+       expiresAt := time.Now().Add(60 * time.Second)
+       signedLocator := SignLocator(pullRequest.Locator, token, expiresAt)
 
        reader, contentLen, _, err := GetContent(signedLocator, keepClient)
        if err != nil {
                return
        }
        if reader == nil {
-               return errors.New(fmt.Sprintf("No reader found for : %s", signedLocator))
+               return fmt.Errorf("No reader found for : %s", signedLocator)
        }
        defer reader.Close()
 
-       read_content, err := ioutil.ReadAll(reader)
+       readContent, err := ioutil.ReadAll(reader)
        if err != nil {
                return err
        }
 
-       if (read_content == nil) || (int64(len(read_content)) != contentLen) {
-               return errors.New(fmt.Sprintf("Content not found for: %s", signedLocator))
+       if (readContent == nil) || (int64(len(readContent)) != contentLen) {
+               return fmt.Errorf("Content not found for: %s", signedLocator)
        }
 
-       err = PutContent(read_content, pullRequest.Locator)
+       err = PutContent(readContent, pullRequest.Locator)
        return
 }
 
@@ -81,19 +80,20 @@ var GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (
        return reader, blocklen, url, err
 }
 
-const ALPHA_NUMERIC = "0123456789abcdefghijklmnopqrstuvwxyz"
+const alphaNumeric = "0123456789abcdefghijklmnopqrstuvwxyz"
 
-func GenerateRandomApiToken() string {
+// GenerateRandomAPIToken generates a random api token
+func GenerateRandomAPIToken() string {
        var bytes = make([]byte, 36)
        rand.Read(bytes)
        for i, b := range bytes {
-               bytes[i] = ALPHA_NUMERIC[b%byte(len(ALPHA_NUMERIC))]
+               bytes[i] = alphaNumeric[b%byte(len(alphaNumeric))]
        }
        return (string(bytes))
 }
 
 // Put block
 var PutContent = func(content []byte, locator string) (err error) {
-       err = PutBlock(content, locator)
+       _, err = PutBlock(content, locator)
        return
 }