17394: Propagate storage classes when writing data via collectionfs.

[arvados.git] / sdk / go / arvados / fs_collection.go

diff --git a/sdk/go/arvados/fs_collection.go b/sdk/go/arvados/fs_collection.go
index b743ab368e33f69a5c1710d63dc410af8a380ffc..4d9db421fc3838b268fdeaeea1b81b9ca1192843 100644 (file)
--- a/sdk/go/arvados/fs_collection.go
+++ b/sdk/go/arvados/fs_collection.go
@@ -42,7 +42,9 @@ type CollectionFileSystem interface {
 
 type collectionFileSystem struct {
        fileSystem
-       uuid string
+       uuid           string
+       replicas       int
+       storageClasses []string
 }
 
 // FileSystem returns a CollectionFileSystem for the collection.
@@ -52,12 +54,16 @@ func (c *Collection) FileSystem(client apiClient, kc keepClient) (CollectionFile
                modTime = time.Now()
        }
        fs := &collectionFileSystem{
-               uuid: c.UUID,
+               uuid:           c.UUID,
+               storageClasses: c.StorageClassesDesired,
                fileSystem: fileSystem{
                        fsBackend: keepBackend{apiClient: client, keepClient: kc},
                        thr:       newThrottle(concurrentWriters),
                },
        }
+       if r := c.ReplicationDesired; r != nil {
+               fs.replicas = *r
+       }
        root := &dirnode{
                fs: fs,
                treenode: treenode{
@@ -321,7 +327,7 @@ func (fn *filenode) seek(startPtr filenodePtr) (ptr filenodePtr) {
 // filenode implements inode.
 type filenode struct {
        parent   inode
-       fs       FileSystem
+       fs       *collectionFileSystem
        fileinfo fileinfo
        segments []segment
        // number of times `segments` has changed in a
@@ -610,7 +616,11 @@ func (fn *filenode) pruneMemSegments() {
                fn.fs.throttle().Acquire()
                go func() {
                        defer close(done)
-                       locator, _, err := fn.FS().PutB(buf)
+                       resp, err := fn.FS().BlockWrite(context.Background(), BlockWriteOptions{
+                               Data:           buf,
+                               Replicas:       fn.fs.replicas,
+                               StorageClasses: fn.fs.storageClasses,
+                       })
                        fn.fs.throttle().Release()
                        fn.Lock()
                        defer fn.Unlock()
@@ -631,7 +641,7 @@ func (fn *filenode) pruneMemSegments() {
                        fn.memsize -= int64(len(buf))
                        fn.segments[idx] = storedSegment{
                                kc:      fn.FS(),
-                               locator: locator,
+                               locator: resp.Locator,
                                size:    len(buf),
                                offset:  0,
                                length:  len(buf),
@@ -748,7 +758,11 @@ func (dn *dirnode) commitBlock(ctx context.Context, refs []fnSegmentRef, bufsize
        go func() {
                defer close(done)
                defer close(errs)
-               locator, _, err := dn.fs.PutB(block)
+               resp, err := dn.fs.BlockWrite(context.Background(), BlockWriteOptions{
+                       Data:           block,
+                       Replicas:       dn.fs.replicas,
+                       StorageClasses: dn.fs.storageClasses,
+               })
                dn.fs.throttle().Release()
                if err != nil {
                        errs <- err
@@ -780,7 +794,7 @@ func (dn *dirnode) commitBlock(ctx context.Context, refs []fnSegmentRef, bufsize
                        data := ref.fn.segments[ref.idx].(*memSegment).buf
                        ref.fn.segments[ref.idx] = storedSegment{
                                kc:      dn.fs,
-                               locator: locator,
+                               locator: resp.Locator,
                                size:    blocksize,
                                offset:  offsets[idx],
                                length:  len(data),