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 := Pull(item.(PullRequest), keepClient)
28 log.Printf("Pull %s success", pullRequest)
30 log.Printf("Pull %s error: %s", pullRequest, err)
36 For each Pull request:
37 Generate a random API token.
38 Generate a permission signature using this token, timestamp ~60 seconds in the future, and desired block hash.
39 Using this token & signature, retrieve the given block.
42 func Pull(pullRequest PullRequest, keepClient keepclient.KeepClient) (err error) {
43 service_roots := make(map[string]string)
44 for _, addr := range pullRequest.Servers {
45 service_roots[addr] = addr
47 keepClient.SetServiceRoots(service_roots)
49 // Generate signature with a random token
50 expires_at := time.Now().Add(60 * time.Second)
51 signedLocator := SignLocator(pullRequest.Locator, GenerateRandomApiToken(), expires_at)
53 reader, contentLen, _, err := GetContent(signedLocator, keepClient)
58 return errors.New(fmt.Sprintf("No reader found for : %s", signedLocator))
62 read_content, err := ioutil.ReadAll(reader)
67 if (read_content == nil) || (int64(len(read_content)) != contentLen) {
68 return errors.New(fmt.Sprintf("Content not found for: %s", signedLocator))
71 err = PutContent(read_content, pullRequest.Locator)
75 // Fetch the content for the given locator using keepclient.
76 var GetContent = func(signedLocator string, keepClient keepclient.KeepClient) (
77 reader io.ReadCloser, contentLength int64, url string, err error) {
78 reader, blocklen, url, err := keepClient.Get(signedLocator)
79 return reader, blocklen, url, err
82 const ALPHA_NUMERIC = "0123456789abcdefghijklmnopqrstuvwxyz"
84 func GenerateRandomApiToken() string {
85 var bytes = make([]byte, 36)
87 for i, b := range bytes {
88 bytes[i] = ALPHA_NUMERIC[b%byte(len(ALPHA_NUMERIC))]
90 return (string(bytes))
94 var PutContent = func(content []byte, locator string) (err error) {
95 err = PutBlock(content, locator)