7 "git.curoverse.com/arvados.git/sdk/go/keepclient"
15 Keepstore initiates pull worker channel goroutine.
16 The channel will process pull list.
17 For each (next) pull request:
18 For each locator listed, execute Pull on the server(s) listed
19 Skip the rest of the servers if no errors
22 func RunPullWorker(pullq *WorkQueue, keepClient *keepclient.KeepClient) {
23 nextItem := pullq.NextItem
24 for item := range nextItem {
25 pullRequest := item.(PullRequest)
26 err := PullItemAndProcess(item.(PullRequest), GenerateRandomApiToken(), keepClient)
27 pullq.ReportDone <- struct{}{}
29 log.Printf("Pull %s success", pullRequest)
31 log.Printf("Pull %s error: %s", pullRequest, err)
37 For each Pull request:
38 Generate a random API token.
39 Generate a permission signature using this token, timestamp ~60 seconds in the future, and desired block hash.
40 Using this token & signature, retrieve the given block.
43 func PullItemAndProcess(pullRequest PullRequest, token string, keepClient *keepclient.KeepClient) (err error) {
44 keepClient.Arvados.ApiToken = token
46 service_roots := make(map[string]string)
47 for _, addr := range pullRequest.Servers {
48 service_roots[addr] = addr
50 keepClient.SetServiceRoots(service_roots, nil, nil)
52 // Generate signature with a random token
53 expires_at := time.Now().Add(60 * time.Second)
54 signedLocator := SignLocator(pullRequest.Locator, token, expires_at)
56 reader, contentLen, _, err := GetContent(signedLocator, keepClient)
61 return errors.New(fmt.Sprintf("No reader found for : %s", signedLocator))
65 read_content, err := ioutil.ReadAll(reader)
70 if (read_content == nil) || (int64(len(read_content)) != contentLen) {
71 return errors.New(fmt.Sprintf("Content not found for: %s", signedLocator))
74 err = PutContent(read_content, pullRequest.Locator)
78 // Fetch the content for the given locator using keepclient.
79 var GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (
80 reader io.ReadCloser, contentLength int64, url string, err error) {
81 reader, blocklen, url, err := keepClient.Get(signedLocator)
82 return reader, blocklen, url, err
85 const ALPHA_NUMERIC = "0123456789abcdefghijklmnopqrstuvwxyz"
87 func GenerateRandomApiToken() string {
88 var bytes = make([]byte, 36)
90 for i, b := range bytes {
91 bytes[i] = ALPHA_NUMERIC[b%byte(len(ALPHA_NUMERIC))]
93 return (string(bytes))
97 var PutContent = func(content []byte, locator string) (err error) {
98 err = PutBlock(content, locator)