type keepClient interface {
ReadAt(locator string, p []byte, off int) (int, error)
PutB(p []byte) (string, int, error)
+ LocalLocator(locator string) (string, error)
}
type apiClient interface {
"encoding/json"
"fmt"
"io"
- "log"
"os"
"path"
"regexp"
}
func (fs *collectionFileSystem) Sync() error {
- log.Printf("cfs.Sync()")
if fs.uuid == "" {
return nil
}
txt, err := fs.MarshalManifest(".")
if err != nil {
- log.Printf("WARNING: (collectionFileSystem)Sync() failed: %s", err)
- return err
+ return fmt.Errorf("sync failed: %s", err)
}
coll := &Collection{
UUID: fs.uuid,
}
err = fs.RequestAndDecode(nil, "PUT", "arvados/v1/collections/"+fs.uuid, fs.UpdateBody(coll), map[string]interface{}{"select": []string{"uuid"}})
if err != nil {
- log.Printf("WARNING: (collectionFileSystem)Sync() failed: %s", err)
+ return fmt.Errorf("sync failed: update %s: %s", fs.uuid, err)
}
- return err
+ return nil
}
func (fs *collectionFileSystem) MarshalManifest(prefix string) (string, error) {
return dn.treenode.Child(name, replace)
}
-// sync flushes in-memory data (for the children with the given names,
-// which must be children of dn) to persistent storage. Caller must
-// have write lock on dn and the named children.
+// sync flushes in-memory data and remote block references (for the
+// children with the given names, which must be children of dn) to
+// local persistent storage. Caller must have write lock on dn and the
+// named children.
func (dn *dirnode) sync(names []string) error {
type shortBlock struct {
fn *filenode
return nil
}
+ localLocator := map[string]string{}
for _, name := range names {
fn, ok := dn.inodes[name].(*filenode)
if !ok {
continue
}
for idx, seg := range fn.segments {
- seg, ok := seg.(*memSegment)
- if !ok {
- continue
- }
- if seg.Len() > maxBlockSize/2 {
- if err := flush([]shortBlock{{fn, idx}}); err != nil {
- return err
+ switch seg := seg.(type) {
+ case storedSegment:
+ loc, ok := localLocator[seg.locator]
+ if !ok {
+ var err error
+ loc, err = dn.fs.LocalLocator(seg.locator)
+ if err != nil {
+ return err
+ }
+ localLocator[seg.locator] = loc
}
- continue
- }
- if pendingLen+seg.Len() > maxBlockSize {
- if err := flush(pending); err != nil {
- return err
+ seg.locator = loc
+ fn.segments[idx] = seg
+ case *memSegment:
+ if seg.Len() > maxBlockSize/2 {
+ if err := flush([]shortBlock{{fn, idx}}); err != nil {
+ return err
+ }
+ continue
}
- pending = nil
- pendingLen = 0
+ if pendingLen+seg.Len() > maxBlockSize {
+ if err := flush(pending); err != nil {
+ return err
+ }
+ pending = nil
+ pendingLen = 0
+ }
+ pending = append(pending, shortBlock{fn, idx})
+ pendingLen += seg.Len()
+ default:
+ panic(fmt.Sprintf("can't sync segment type %T", seg))
}
- pending = append(pending, shortBlock{fn, idx})
- pendingLen += seg.Len()
}
}
return flush(pending)
}
-// caller must have read lock.
+// caller must have write lock.
func (dn *dirnode) marshalManifest(prefix string) (string, error) {
var streamLen int64
type filepart struct {
import (
"bytes"
"crypto/md5"
+ "crypto/sha1"
"errors"
"fmt"
"io"
"os"
"regexp"
"runtime"
+ "strings"
"sync"
"testing"
"time"
var _ = check.Suite(&CollectionFSSuite{})
type keepClientStub struct {
- blocks map[string][]byte
+ blocks map[string][]byte
+ refreshable map[string]bool
sync.RWMutex
}
return locator, 1, nil
}
+var localOrRemoteSignature = regexp.MustCompile(`\+[AR][^+]*`)
+
+func (kcs *keepClientStub) LocalLocator(locator string) (string, error) {
+ kcs.Lock()
+ defer kcs.Unlock()
+ if strings.Contains(locator, "+R") {
+ if len(locator) < 32 {
+ return "", fmt.Errorf("bad locator: %q", locator)
+ }
+ if _, ok := kcs.blocks[locator[:32]]; !ok && !kcs.refreshable[locator[:32]] {
+ return "", fmt.Errorf("kcs.refreshable[%q]==false", locator)
+ }
+ }
+ fakeSig := fmt.Sprintf("+A%x@%x", sha1.Sum(nil), time.Now().Add(time.Hour*24*14).Unix())
+ return localOrRemoteSignature.ReplaceAllLiteralString(locator, fakeSig), nil
+}
+
type CollectionFSSuite struct {
client *Client
coll Collection
fs CollectionFileSystem
- kc keepClient
+ kc *keepClientStub
}
func (s *CollectionFSSuite) SetUpTest(c *check.C) {
checkSize(11)
}
+func (s *CollectionFSSuite) TestMarshalCopiesRemoteBlocks(c *check.C) {
+ foo := "foo"
+ bar := "bar"
+ hash := map[string]string{
+ foo: fmt.Sprintf("%x", md5.Sum([]byte(foo))),
+ bar: fmt.Sprintf("%x", md5.Sum([]byte(bar))),
+ }
+
+ fs, err := (&Collection{
+ ManifestText: ". " + hash[foo] + "+3+Rzaaaa-foo@bab " + hash[bar] + "+3+A12345@ffffff 0:2:fo.txt 2:4:obar.txt\n",
+ }).FileSystem(s.client, s.kc)
+ c.Assert(err, check.IsNil)
+ manifest, err := fs.MarshalManifest(".")
+ c.Check(manifest, check.Equals, "")
+ c.Check(err, check.NotNil)
+
+ s.kc.refreshable = map[string]bool{hash[bar]: true}
+
+ for _, sigIn := range []string{"Rzaaaa-foo@bab", "A12345@abcde"} {
+ fs, err = (&Collection{
+ ManifestText: ". " + hash[foo] + "+3+A12345@fffff " + hash[bar] + "+3+" + sigIn + " 0:2:fo.txt 2:4:obar.txt\n",
+ }).FileSystem(s.client, s.kc)
+ c.Assert(err, check.IsNil)
+ manifest, err := fs.MarshalManifest(".")
+ c.Check(err, check.IsNil)
+ // Both blocks should now have +A signatures.
+ c.Check(manifest, check.Matches, `.*\+A.* .*\+A.*\n`)
+ c.Check(manifest, check.Not(check.Matches), `.*\+R.*\n`)
+ }
+}
+
func (s *CollectionFSSuite) TestMarshalSmallBlocks(c *check.C) {
maxBlockSize = 8
defer func() { maxBlockSize = 2 << 26 }()
f, err := s.fs.OpenFile("/dir1/foo", os.O_RDWR, 0)
c.Assert(err, check.IsNil)
defer f.Close()
- for i := 0; i < 6502; i++ {
+ for i := 0; i < 1024; i++ {
r := rand.Uint32()
switch {
case r%11 == 0:
}
}
-func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, int64, string, error) {
+func (kc *KeepClient) getOrHead(method string, locator string, header http.Header) (io.ReadCloser, int64, string, http.Header, error) {
if strings.HasPrefix(locator, "d41d8cd98f00b204e9800998ecf8427e+0") {
- return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil
+ return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil, nil
}
reqid := kc.getRequestID()
errs = append(errs, fmt.Sprintf("%s: %v", url, err))
continue
}
- req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
- req.Header.Add("X-Request-Id", reqid)
+ for k, v := range header {
+ req.Header[k] = append([]string(nil), v...)
+ }
+ if req.Header.Get("Authorization") == "" {
+ req.Header.Set("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
+ }
+ if req.Header.Get("X-Request-Id") == "" {
+ req.Header.Set("X-Request-Id", reqid)
+ }
resp, err := kc.httpClient().Do(req)
if err != nil {
// Probably a network error, may be transient,
if expectLength < 0 {
if resp.ContentLength < 0 {
resp.Body.Close()
- return nil, 0, "", fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator)
+ return nil, 0, "", nil, fmt.Errorf("error reading %q: no size hint, no Content-Length header in response", locator)
}
expectLength = resp.ContentLength
} else if resp.ContentLength >= 0 && expectLength != resp.ContentLength {
resp.Body.Close()
- return nil, 0, "", fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength)
+ return nil, 0, "", nil, fmt.Errorf("error reading %q: size hint %d != Content-Length %d", locator, expectLength, resp.ContentLength)
}
// Success
if method == "GET" {
Reader: resp.Body,
Hash: md5.New(),
Check: locator[0:32],
- }, expectLength, url, nil
+ }, expectLength, url, resp.Header, nil
} else {
resp.Body.Close()
- return nil, expectLength, url, nil
+ return nil, expectLength, url, resp.Header, nil
}
}
serversToTry = retryList
isTemp: len(serversToTry) > 0,
}}
}
- return nil, 0, "", err
+ return nil, 0, "", nil, err
+}
+
+// LocalLocator returns a locator equivalent to the one supplied, but
+// with a valid signature from the local cluster. If the given locator
+// already has a local signature, it is returned unchanged.
+func (kc *KeepClient) LocalLocator(locator string) (string, error) {
+ if !strings.Contains(locator, "+R") {
+ // Either it has +A, or it's unsigned and we assume
+ // it's a local locator on a site with signatures
+ // disabled.
+ return locator, nil
+ }
+ sighdr := fmt.Sprintf("local, time=%s", time.Now().UTC().Format(time.RFC3339))
+ _, _, url, hdr, err := kc.getOrHead("HEAD", locator, http.Header{"X-Keep-Signature": []string{sighdr}})
+ if err != nil {
+ return "", err
+ }
+ loc := hdr.Get("X-Keep-Locator")
+ if loc == "" {
+ return "", fmt.Errorf("missing X-Keep-Locator header in HEAD response from %s", url)
+ }
+ return loc, nil
}
// Get() retrieves a block, given a locator. Returns a reader, the
// reader returned by this method will return a BadChecksum error
// instead of EOF.
func (kc *KeepClient) Get(locator string) (io.ReadCloser, int64, string, error) {
- return kc.getOrHead("GET", locator)
+ rdr, size, url, _, err := kc.getOrHead("GET", locator, nil)
+ return rdr, size, url, err
}
// ReadAt() retrieves a portion of block from the cache if it's
// Returns the data size (content length) reported by the Keep service
// and the URI reporting the data size.
func (kc *KeepClient) Ask(locator string) (int64, string, error) {
- _, size, url, err := kc.getOrHead("HEAD", locator)
+ _, size, url, _, err := kc.getOrHead("HEAD", locator, nil)
return size, url, err
}
c.Check(n, Equals, int64(len(content)))
c.Check(url2, Matches, fmt.Sprintf("http://localhost:\\d+/%s", hash))
}
+ {
+ loc, err := kc.LocalLocator(hash)
+ c.Check(err, Equals, nil)
+ c.Assert(len(loc) >= 32, Equals, true)
+ c.Check(loc[:32], Equals, hash[:32])
+ }
+ {
+ content := []byte("the perth county conspiracy")
+ loc, err := kc.LocalLocator(fmt.Sprintf("%x+%d+Rzaaaa-abcde@12345", md5.Sum(content), len(content)))
+ c.Check(loc, Equals, "")
+ c.Check(err, ErrorMatches, `.*HEAD .*\+R.*`)
+ c.Check(err, ErrorMatches, `.*HTTP 400.*`)
+ }
}
type StubProxyHandler struct {
PutB(buf []byte) (string, int, error)
ReadAt(locator string, p []byte, off int) (int, error)
ManifestFileReader(m manifest.Manifest, filename string) (arvados.File, error)
+ LocalLocator(locator string) (string, error)
ClearBlockCache()
}
if err != nil {
return err
}
+ if n := len(regexp.MustCompile(` [0-9a-f]+\+\S*\+R`).FindAllStringIndex(txt, -1)); n > 0 {
+ runner.CrunchLog.Printf("Copying %d data blocks from remote input collections...", n)
+ fs, err := (&arvados.Collection{ManifestText: txt}).FileSystem(runner.client, runner.Kc)
+ if err != nil {
+ return err
+ }
+ txt, err = fs.MarshalManifest(".")
+ if err != nil {
+ return err
+ }
+ }
var resp arvados.Collection
err = runner.ArvClient.Create("collections", arvadosclient.Dict{
"ensure_unique_name": true,
return nil
}
+func (client *KeepTestClient) LocalLocator(locator string) (string, error) {
+ return locator, nil
+}
+
func (client *KeepTestClient) PutB(buf []byte) (string, int, error) {
client.Content = buf
return fmt.Sprintf("%x+%d", md5.Sum(buf), len(buf)), len(buf), nil
return "", 0, errors.New("KeepError")
}
+func (*KeepErrorTestClient) LocalLocator(string) (string, error) {
+ return "", errors.New("KeepError")
+}
+
type KeepReadErrorTestClient struct {
KeepTestClient
}
remoteID := part[1:6]
remote, ok := cluster.RemoteClusters[remoteID]
if !ok {
- http.Error(w, "remote cluster not configured", http.StatusBadGateway)
+ http.Error(w, "remote cluster not configured", http.StatusBadRequest)
return
}
kc, err := rp.remoteClient(remoteID, remote, token)
projects / arvados.git / commitdiff