Merge branch 'master' into 7241-azure-blob-volume

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

diff --git a/services/keepstore/pull_worker.go b/services/keepstore/pull_worker.go
index dced531bf5fb08ac8bd808818ca8bf021128de31..2626d4bf68e1594f394ad4539f0f32a90fe00339 100644 (file)
--- a/services/keepstore/pull_worker.go
+++ b/services/keepstore/pull_worker.go
@@ -8,55 +8,51 @@ import (
        "io"
        "io/ioutil"
        "log"
-       "os"
        "time"
 )
 
-var keepClient keepclient.KeepClient
-
-/*
-       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
-*/
-func RunPullWorker(nextItem <-chan interface{}, kc keepclient.KeepClient) {
-       keepClient = kc
+// 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 {
-               Pull(item.(PullRequest))
-       }
-}
-
-/*
-       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 Pull(pullRequest PullRequest) (err error) {
-       defer func() {
+               pullRequest := item.(PullRequest)
+               err := PullItemAndProcess(item.(PullRequest), GenerateRandomAPIToken(), keepClient)
+               pullq.DoneItem <- struct{}{}
                if err == nil {
                        log.Printf("Pull %s success", pullRequest)
                } else {
                        log.Printf("Pull %s error: %s", pullRequest, err)
                }
-       }()
+       }
+}
+
+// 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)
+       keepClient.SetServiceRoots(serviceRoots, nil, nil)
 
        // Generate signature with a random token
-       PermissionSecret = []byte(os.Getenv("ARVADOS_API_TOKEN"))
-       expires_at := time.Now().Add(60 * time.Second)
-       signedLocator := SignLocator(pullRequest.Locator, GenerateRandomApiToken(), expires_at)
+       expiresAt := time.Now().Add(60 * time.Second)
+       signedLocator := SignLocator(pullRequest.Locator, token, expiresAt)
 
-       reader, contentLen, _, err := GetContent(pullRequest.Locator, signedLocator)
+       reader, contentLen, _, err := GetContent(signedLocator, keepClient)
        if err != nil {
                return
        }
@@ -65,38 +61,40 @@ func Pull(pullRequest PullRequest) (err error) {
        }
        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) {
+       if (readContent == nil) || (int64(len(readContent)) != contentLen) {
                return errors.New(fmt.Sprintf("Content not found for: %s", signedLocator))
        }
 
-       err = PutContent(read_content, pullRequest.Locator)
+       err = PutContent(readContent, pullRequest.Locator)
        return
 }
 
 // Fetch the content for the given locator using keepclient.
-var GetContent = func(locator string, signedLocator string) (reader io.ReadCloser, contentLength int64, url string, err error) {
+var GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (
+       reader io.ReadCloser, contentLength int64, url string, err error) {
        reader, blocklen, url, err := keepClient.Get(signedLocator)
        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
 }