7 "git.curoverse.com/arvados.git/sdk/go/keepclient"
15 var keepClient keepclient.KeepClient
18 Keepstore initiates pull worker channel goroutine.
19 The channel will process pull list.
20 For each (next) pull request:
21 For each locator listed, execute Pull on the server(s) listed
22 Skip the rest of the servers if no errors
25 func RunPullWorker(pullq *WorkQueue, kc keepclient.KeepClient) {
26 nextItem := pullq.NextItem
28 for item := range nextItem {
29 Pull(item.(PullRequest))
34 For each Pull request:
35 Generate a random API token.
36 Generate a permission signature using this token, timestamp ~60 seconds in the future, and desired block hash.
37 Using this token & signature, retrieve the given block.
40 func Pull(pullRequest PullRequest) (err error) {
43 log.Printf("Pull %s success", pullRequest)
45 log.Printf("Pull %s error: %s", pullRequest, err)
49 service_roots := make(map[string]string)
50 for _, addr := range pullRequest.Servers {
51 service_roots[addr] = addr
53 keepClient.SetServiceRoots(service_roots)
55 // Generate signature with a random token
56 PermissionSecret = []byte(os.Getenv("ARVADOS_API_TOKEN"))
57 expires_at := time.Now().Add(60 * time.Second)
58 signedLocator := SignLocator(pullRequest.Locator, GenerateRandomApiToken(), expires_at)
60 reader, contentLen, _, err := GetContent(pullRequest.Locator, signedLocator)
65 return errors.New(fmt.Sprintf("No reader found for : %s", signedLocator))
69 read_content, err := ioutil.ReadAll(reader)
74 if (read_content == nil) || (int64(len(read_content)) != contentLen) {
75 return errors.New(fmt.Sprintf("Content not found for: %s", signedLocator))
78 err = PutContent(read_content, pullRequest.Locator)
82 // Fetch the content for the given locator using keepclient.
83 var GetContent = func(locator string, signedLocator string) (reader io.ReadCloser, contentLength int64, url string, err error) {
84 reader, blocklen, url, err := keepClient.Get(signedLocator)
85 return reader, blocklen, url, err
88 const ALPHA_NUMERIC = "0123456789abcdefghijklmnopqrstuvwxyz"
90 func GenerateRandomApiToken() string {
91 var bytes = make([]byte, 36)
93 for i, b := range bytes {
94 bytes[i] = ALPHA_NUMERIC[b%byte(len(ALPHA_NUMERIC))]
96 return (string(bytes))
100 var PutContent = func(content []byte, locator string) (err error) {
101 err = PutBlock(content, locator)