X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/cb9fbffa7f480dae5f17eb44f27d0b3523da0f0a..601764a10575232ada34a7c0dc5ec61195094e7f:/services/keepstore/pull_worker.go diff --git a/services/keepstore/pull_worker.go b/services/keepstore/pull_worker.go index 2342fd2256..42b5d5889d 100644 --- a/services/keepstore/pull_worker.go +++ b/services/keepstore/pull_worker.go @@ -1,118 +1,110 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( + "context" "crypto/rand" - "errors" "fmt" - "git.curoverse.com/arvados.git/sdk/go/arvadosclient" - "git.curoverse.com/arvados.git/sdk/go/keepclient" + "io" "io/ioutil" - "log" - "os" - "strconv" "time" -) -var arv arvadosclient.ArvadosClient -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{}) { - var err error - arv, err = arvadosclient.MakeArvadosClient() - if err != nil { - log.Fatalf("Error setting up arvados client %s", err.Error()) - } - arv.ApiToken = os.Getenv("ARVADOS_API_TOKEN") - - keepClient, err = keepclient.MakeKeepClient(&arv) - if err != nil { - log.Fatalf("Error setting up keep client %s", err.Error()) - } + "git.curoverse.com/arvados.git/sdk/go/keepclient" +) - for item := range nextItem { - pullReq := item.(PullRequest) - for _, addr := range pullReq.Servers { - err := Pull(addr, pullReq.Locator) - if err == nil { - break - } +// RunPullWorker receives PullRequests from pullq, invokes +// PullItemAndProcess on each one. After each PR, it logs a message +// indicating whether the pull was successful. +func RunPullWorker(pullq *WorkQueue, keepClient *keepclient.KeepClient) { + for item := range pullq.NextItem { + pr := item.(PullRequest) + err := PullItemAndProcess(pr, keepClient) + pullq.DoneItem <- struct{}{} + if err == nil { + log.Printf("Pull %s success", pr) + } else { + log.Printf("Pull %s error: %s", pr, err) } } } -/* - 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(addr string, locator string) (err error) { - log.Printf("Pull %s/%s starting", addr, locator) - - defer func() { - if err == nil { - log.Printf("Pull %s/%s success", addr, locator) - } else { - log.Printf("Pull %s/%s error: %s", addr, locator, err) +// PullItemAndProcess executes a pull request by retrieving the +// specified block from one of the specified servers, and storing it +// on a local volume. +// +// If the PR specifies a non-blank mount UUID, PullItemAndProcess will +// only attempt to write the data to the corresponding +// volume. Otherwise it writes to any local volume, as a PUT request +// would. +func PullItemAndProcess(pullRequest PullRequest, keepClient *keepclient.KeepClient) error { + var vol Volume + if uuid := pullRequest.MountUUID; uuid != "" { + vol = KeepVM.Lookup(pullRequest.MountUUID, true) + if vol == nil { + return fmt.Errorf("pull req has nonexistent mount: %v", pullRequest) } - }() + } - service_roots := make(map[string]string) - service_roots[locator] = addr - keepClient.SetServiceRoots(service_roots) + keepClient.Arvados.ApiToken = randomToken - read_content, err := GetContent(addr, locator) - log.Print(read_content, err) - if err != nil { - return + serviceRoots := make(map[string]string) + for _, addr := range pullRequest.Servers { + serviceRoots[addr] = addr } + keepClient.SetServiceRoots(serviceRoots, nil, nil) - err = PutBlock(read_content, locator) - return -} - -// Fetch the content for the given locator using keepclient. -var GetContent = func(addr string, locator string) ([]byte, error) { // Generate signature with a random token - expires_at := time.Now().Unix() + 60 // now + 1 min in seconds - hints := "+A" + GenerateRandomApiToken() + "@" + strconv.FormatInt(expires_at, 16) - signature := keepclient.MakeLocator2(locator, hints) + expiresAt := time.Now().Add(60 * time.Second) + signedLocator := SignLocator(pullRequest.Locator, randomToken, expiresAt) - reader, blocklen, _, err := keepClient.AuthorizedGet(locator, signature.Signature, signature.Timestamp) - defer reader.Close() + reader, contentLen, _, err := GetContent(signedLocator, keepClient) if err != nil { - return nil, err + return err + } + if reader == nil { + return fmt.Errorf("No reader found for : %s", signedLocator) } + defer reader.Close() - read_content, err := ioutil.ReadAll(reader) - log.Print(read_content, err) + readContent, err := ioutil.ReadAll(reader) if err != nil { - return nil, err + return err } - if (read_content == nil) || (int64(len(read_content)) != blocklen) { - return nil, errors.New(fmt.Sprintf("Content not found for: %s/%s", addr, locator)) + if (readContent == nil) || (int64(len(readContent)) != contentLen) { + return fmt.Errorf("Content not found for: %s", signedLocator) } - return read_content, nil + writePulledBlock(vol, readContent, pullRequest.Locator) + return nil } -const ALPHA_NUMERIC = "0123456789abcdefghijklmnopqrstuvwxyz" +// Fetch the content for the given locator using keepclient. +var GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (io.ReadCloser, int64, string, error) { + return keepClient.Get(signedLocator) +} + +var writePulledBlock = func(volume Volume, data []byte, locator string) { + var err error + if volume != nil { + err = volume.Put(context.Background(), locator, data) + } else { + _, err = PutBlock(context.Background(), data, locator) + } + if err != nil { + log.Printf("error writing pulled block %q: %s", locator, err) + } +} -func GenerateRandomApiToken() string { +var randomToken = func() string { + const alphaNumeric = "0123456789abcdefghijklmnopqrstuvwxyz" 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)) -} +}()