11 "git.curoverse.com/arvados.git/sdk/go/keepclient"
13 log "github.com/Sirupsen/logrus"
16 // RunPullWorker is used by Keepstore to initiate pull worker channel goroutine.
17 // The channel will process pull list.
18 // For each (next) pull request:
19 // For each locator listed, execute Pull on the server(s) listed
20 // Skip the rest of the servers if no errors
23 func RunPullWorker(pullq *WorkQueue, keepClient *keepclient.KeepClient) {
24 nextItem := pullq.NextItem
25 for item := range nextItem {
26 pullRequest := item.(PullRequest)
27 err := PullItemAndProcess(item.(PullRequest), keepClient)
28 pullq.DoneItem <- struct{}{}
30 log.Printf("Pull %s success", pullRequest)
32 log.Printf("Pull %s error: %s", pullRequest, err)
37 // PullItemAndProcess pulls items from PullQueue and processes them.
38 // For each Pull request:
39 // Generate a random API token.
40 // Generate a permission signature using this token, timestamp ~60 seconds in the future, and desired block hash.
41 // Using this token & signature, retrieve the given block.
44 func PullItemAndProcess(pullRequest PullRequest, keepClient *keepclient.KeepClient) error {
46 if uuid := pullRequest.MountUUID; uuid != "" {
47 vol = KeepVM.Lookup(pullRequest.MountUUID, true)
49 return fmt.Errorf("pull req has nonexistent mount: %v", pullRequest)
53 keepClient.Arvados.ApiToken = randomToken
55 serviceRoots := make(map[string]string)
56 for _, addr := range pullRequest.Servers {
57 serviceRoots[addr] = addr
59 keepClient.SetServiceRoots(serviceRoots, nil, nil)
61 // Generate signature with a random token
62 expiresAt := time.Now().Add(60 * time.Second)
63 signedLocator := SignLocator(pullRequest.Locator, randomToken, expiresAt)
65 reader, contentLen, _, err := GetContent(signedLocator, keepClient)
70 return fmt.Errorf("No reader found for : %s", signedLocator)
74 readContent, err := ioutil.ReadAll(reader)
79 if (readContent == nil) || (int64(len(readContent)) != contentLen) {
80 return fmt.Errorf("Content not found for: %s", signedLocator)
83 writePulledBlock(vol, readContent, pullRequest.Locator)
87 // Fetch the content for the given locator using keepclient.
88 var GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (io.ReadCloser, int64, string, error) {
89 return keepClient.Get(signedLocator)
92 var writePulledBlock = func(volume Volume, data []byte, locator string) {
95 err = volume.Put(context.Background(), locator, data)
97 _, err = PutBlock(context.Background(), data, locator)
100 log.Printf("error writing pulled block %q: %s", locator, err)
104 var randomToken = func() string {
105 const alphaNumeric = "0123456789abcdefghijklmnopqrstuvwxyz"
106 var bytes = make([]byte, 36)
108 for i, b := range bytes {
109 bytes[i] = alphaNumeric[b%byte(len(alphaNumeric))]
111 return (string(bytes))