lib/crunchrun/copier.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package crunchrun
   6
   7 import (
   8         "encoding/json"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "os"
  13         "path/filepath"
  14         "sort"
  15         "strings"
  16
  17         "git.arvados.org/arvados.git/sdk/go/arvados"
  18         "git.arvados.org/arvados.git/sdk/go/keepclient"
  19         "git.arvados.org/arvados.git/sdk/go/manifest"
  20 )
  21
  22 type printfer interface {
  23         Printf(string, ...interface{})
  24 }
  25
  26 var errTooManySymlinks = errors.New("too many symlinks, or symlink cycle")
  27
  28 const limitFollowSymlinks = 10
  29
  30 type filetodo struct {
  31         src  string
  32         dst  string
  33         size int64
  34 }
  35
  36 // copier copies data from a finished container's output path to a new
  37 // Arvados collection.
  38 //
  39 // Regular files (and symlinks to regular files) in hostOutputDir are
  40 // copied from the local filesystem.
  41 //
  42 // Symlinks to mounted collections, and any collections mounted under
  43 // ctrOutputDir, are copied by transforming the relevant parts of the
  44 // existing manifests, without moving any data around.
  45 //
  46 // Symlinks to other parts of the container's filesystem result in
  47 // errors.
  48 //
  49 // Use:
  50 //
  51 //      manifest, err := (&copier{...}).Copy()
  52 type copier struct {
  53         client        *arvados.Client
  54         arvClient     IArvadosClient
  55         keepClient    IKeepClient
  56         hostOutputDir string
  57         ctrOutputDir  string
  58         bindmounts    map[string]bindmount
  59         mounts        map[string]arvados.Mount
  60         secretMounts  map[string]arvados.Mount
  61         logger        printfer
  62
  63         dirs     []string
  64         files    []filetodo
  65         manifest string
  66
  67         manifestCache map[string]*manifest.Manifest
  68 }
  69
  70 // Copy copies data as needed, and returns a new manifest.
  71 func (cp *copier) Copy() (string, error) {
  72         err := cp.walkMount("", cp.ctrOutputDir, limitFollowSymlinks, true)
  73         if err != nil {
  74                 return "", fmt.Errorf("error scanning files to copy to output: %v", err)
  75         }
  76         fs, err := (&arvados.Collection{ManifestText: cp.manifest}).FileSystem(cp.client, cp.keepClient)
  77         if err != nil {
  78                 return "", fmt.Errorf("error creating Collection.FileSystem: %v", err)
  79         }
  80         for _, d := range cp.dirs {
  81                 err = fs.Mkdir(d, 0777)
  82                 if err != nil && err != os.ErrExist {
  83                         return "", fmt.Errorf("error making directory %q in output collection: %v", d, err)
  84                 }
  85         }
  86         var unflushed int64
  87         var lastparentdir string
  88         for _, f := range cp.files {
  89                 // If a dir has just had its last file added, do a
  90                 // full Flush. Otherwise, do a partial Flush (write
  91                 // full-size blocks, but leave the last short block
  92                 // open so f's data can be packed with it).
  93                 dir, _ := filepath.Split(f.dst)
  94                 if dir != lastparentdir || unflushed > keepclient.BLOCKSIZE {
  95                         if err := fs.Flush("/"+lastparentdir, dir != lastparentdir); err != nil {
  96                                 return "", fmt.Errorf("error flushing output collection file data: %v", err)
  97                         }
  98                         unflushed = 0
  99                 }
 100                 lastparentdir = dir
 101
 102                 n, err := cp.copyFile(fs, f)
 103                 if err != nil {
 104                         return "", fmt.Errorf("error copying file %q into output collection: %v", f, err)
 105                 }
 106                 unflushed += n
 107         }
 108         return fs.MarshalManifest(".")
 109 }
 110
 111 func (cp *copier) copyFile(fs arvados.CollectionFileSystem, f filetodo) (int64, error) {
 112         cp.logger.Printf("copying %q (%d bytes)", strings.TrimLeft(f.dst, "/"), f.size)
 113         dst, err := fs.OpenFile(f.dst, os.O_CREATE|os.O_WRONLY, 0666)
 114         if err != nil {
 115                 return 0, err
 116         }
 117         src, err := os.Open(f.src)
 118         if err != nil {
 119                 dst.Close()
 120                 return 0, err
 121         }
 122         defer src.Close()
 123         n, err := io.Copy(dst, src)
 124         if err != nil {
 125                 dst.Close()
 126                 return n, err
 127         }
 128         return n, dst.Close()
 129 }
 130
 131 // Append to cp.manifest, cp.files, and cp.dirs so as to copy src (an
 132 // absolute path in the container's filesystem) to dest (an absolute
 133 // path in the output collection, or "" for output root).
 134 //
 135 // src must be (or be a descendant of) a readonly "collection" mount,
 136 // a writable collection mounted at ctrOutputPath, or a "tmp" mount.
 137 //
 138 // If walkMountsBelow is true, include contents of any collection
 139 // mounted below src as well.
 140 func (cp *copier) walkMount(dest, src string, maxSymlinks int, walkMountsBelow bool) error {
 141         // srcRoot, srcMount indicate the innermost mount that
 142         // contains src.
 143         var srcRoot string
 144         var srcMount arvados.Mount
 145         for root, mnt := range cp.mounts {
 146                 if len(root) > len(srcRoot) && strings.HasPrefix(src+"/", root+"/") {
 147                         srcRoot, srcMount = root, mnt
 148                 }
 149         }
 150         for root := range cp.secretMounts {
 151                 if len(root) > len(srcRoot) && strings.HasPrefix(src+"/", root+"/") {
 152                         // Silently omit secrets, and symlinks to
 153                         // secrets.
 154                         return nil
 155                 }
 156         }
 157         if srcRoot == "" {
 158                 return fmt.Errorf("cannot output file %q: not in any mount", src)
 159         }
 160
 161         // srcRelPath is the path to the file/dir we are trying to
 162         // copy, relative to its mount point -- ".", "./foo.txt", ...
 163         srcRelPath := filepath.Join(".", srcMount.Path, src[len(srcRoot):])
 164
 165         // outputRelPath is the path relative in the output directory
 166         // that corresponds to the path in the output collection where
 167         // the file will go, for logging
 168         var outputRelPath = ""
 169         if strings.HasPrefix(src, cp.ctrOutputDir) {
 170                 outputRelPath = strings.TrimPrefix(src[len(cp.ctrOutputDir):], "/")
 171         }
 172         if outputRelPath == "" {
 173                 // blank means copy a whole directory, so replace it
 174                 // with a wildcard to make it a little clearer what's
 175                 // going on since outputRelPath is only used for logging
 176                 outputRelPath = "*"
 177         }
 178
 179         switch {
 180         case srcMount.ExcludeFromOutput:
 181         case srcMount.Kind == "tmp":
 182                 // Handle by walking the host filesystem.
 183                 return cp.walkHostFS(dest, src, maxSymlinks, walkMountsBelow)
 184         case srcMount.Kind != "collection":
 185                 return fmt.Errorf("%q: unsupported mount %q in output (kind is %q)", src, srcRoot, srcMount.Kind)
 186         case !srcMount.Writable:
 187                 cp.logger.Printf("copying %q from %v/%v", outputRelPath, srcMount.PortableDataHash, strings.TrimPrefix(srcRelPath, "./"))
 188                 mft, err := cp.getManifest(srcMount.PortableDataHash)
 189                 if err != nil {
 190                         return err
 191                 }
 192                 cp.manifest += mft.Extract(srcRelPath, dest).Text
 193         default:
 194                 cp.logger.Printf("copying %q", outputRelPath)
 195                 hostRoot, err := cp.hostRoot(srcRoot)
 196                 if err != nil {
 197                         return err
 198                 }
 199                 f, err := os.Open(filepath.Join(hostRoot, ".arvados#collection"))
 200                 if err != nil {
 201                         return err
 202                 }
 203                 defer f.Close()
 204                 var coll arvados.Collection
 205                 err = json.NewDecoder(f).Decode(&coll)
 206                 if err != nil {
 207                         return err
 208                 }
 209                 mft := manifest.Manifest{Text: coll.ManifestText}
 210                 cp.manifest += mft.Extract(srcRelPath, dest).Text
 211         }
 212         if walkMountsBelow {
 213                 return cp.walkMountsBelow(dest, src)
 214         }
 215         return nil
 216 }
 217
 218 func (cp *copier) walkMountsBelow(dest, src string) error {
 219         for mnt, mntinfo := range cp.mounts {
 220                 if !strings.HasPrefix(mnt, src+"/") {
 221                         continue
 222                 }
 223                 if cp.copyRegularFiles(mntinfo) {
 224                         // These got copied into the nearest parent
 225                         // mount as regular files during setup, so
 226                         // they get copied as regular files when we
 227                         // process the parent. Output will reflect any
 228                         // changes and deletions done by the
 229                         // container.
 230                         continue
 231                 }
 232                 // Example: we are processing dest=/foo src=/mnt1/dir1
 233                 // (perhaps we followed a symlink /outdir/foo ->
 234                 // /mnt1/dir1). Caller has already processed the
 235                 // collection mounted at /mnt1, but now we find that
 236                 // /mnt1/dir1/mnt2 is also a mount, so we need to copy
 237                 // src=/mnt1/dir1/mnt2 to dest=/foo/mnt2.
 238                 //
 239                 // We handle all descendants of /mnt1/dir1 in this
 240                 // loop instead of using recursion:
 241                 // /mnt1/dir1/mnt2/mnt3 is a child of both /mnt1 and
 242                 // /mnt1/dir1/mnt2, but we only want to walk it
 243                 // once. (This simplification is safe because mounted
 244                 // collections cannot contain symlinks.)
 245                 err := cp.walkMount(dest+mnt[len(src):], mnt, 0, false)
 246                 if err != nil {
 247                         return err
 248                 }
 249         }
 250         return nil
 251 }
 252
 253 // Add entries to cp.dirs and cp.files so as to copy src (an absolute
 254 // path in the container's filesystem which corresponds to a real file
 255 // or directory in cp.hostOutputDir) to dest (an absolute path in the
 256 // output collection, or "" for output root).
 257 //
 258 // Always follow symlinks.
 259 //
 260 // If includeMounts is true, include mounts at and below src.
 261 // Otherwise, skip them.
 262 func (cp *copier) walkHostFS(dest, src string, maxSymlinks int, includeMounts bool) error {
 263         if includeMounts {
 264                 err := cp.walkMountsBelow(dest, src)
 265                 if err != nil {
 266                         return err
 267                 }
 268         }
 269
 270         hostsrc := cp.hostOutputDir + src[len(cp.ctrOutputDir):]
 271
 272         // If src is a symlink, walk its target.
 273         fi, err := os.Lstat(hostsrc)
 274         if err != nil {
 275                 return fmt.Errorf("lstat %q: %s", src, err)
 276         }
 277         if fi.Mode()&os.ModeSymlink != 0 {
 278                 if maxSymlinks < 0 {
 279                         return errTooManySymlinks
 280                 }
 281                 target, err := os.Readlink(hostsrc)
 282                 if err != nil {
 283                         return fmt.Errorf("readlink %q: %s", src, err)
 284                 }
 285                 if !strings.HasPrefix(target, "/") {
 286                         target = filepath.Join(filepath.Dir(src), target)
 287                 }
 288                 return cp.walkMount(dest, target, maxSymlinks-1, true)
 289         }
 290
 291         // If src is a regular directory, append it to cp.dirs and
 292         // walk each of its children. (If there are no children,
 293         // create an empty file "dest/.keep".)
 294         if fi.Mode().IsDir() {
 295                 if dest != "" {
 296                         cp.dirs = append(cp.dirs, dest)
 297                 }
 298                 dir, err := os.Open(hostsrc)
 299                 if err != nil {
 300                         return fmt.Errorf("open %q: %s", src, err)
 301                 }
 302                 names, err := dir.Readdirnames(-1)
 303                 dir.Close()
 304                 if err != nil {
 305                         return fmt.Errorf("readdirnames %q: %s", src, err)
 306                 }
 307                 if len(names) == 0 {
 308                         if dest != "" {
 309                                 cp.files = append(cp.files, filetodo{
 310                                         src: os.DevNull,
 311                                         dst: dest + "/.keep",
 312                                 })
 313                         }
 314                         return nil
 315                 }
 316                 sort.Strings(names)
 317                 for _, name := range names {
 318                         dest, src := dest+"/"+name, src+"/"+name
 319                         if _, isSecret := cp.secretMounts[src]; isSecret {
 320                                 continue
 321                         }
 322                         if mntinfo, isMount := cp.mounts[src]; isMount && !cp.copyRegularFiles(mntinfo) {
 323                                 // If a regular file/dir somehow
 324                                 // exists at a path that's also a
 325                                 // mount target, ignore the file --
 326                                 // the mount has already been included
 327                                 // with walkMountsBelow().
 328                                 //
 329                                 // (...except mount types that are
 330                                 // handled as regular files.)
 331                                 continue
 332                         }
 333                         err = cp.walkHostFS(dest, src, maxSymlinks, false)
 334                         if err != nil {
 335                                 return err
 336                         }
 337                 }
 338                 return nil
 339         }
 340
 341         // If src is a regular file, append it to cp.files.
 342         if fi.Mode().IsRegular() {
 343                 cp.files = append(cp.files, filetodo{
 344                         src:  hostsrc,
 345                         dst:  dest,
 346                         size: fi.Size(),
 347                 })
 348                 return nil
 349         }
 350         cp.logger.Printf("Skipping unsupported file type (mode %o) in output dir: %q", fi.Mode(), src)
 351         return nil
 352 }
 353
 354 // Return the host path that was mounted at the given path in the
 355 // container.
 356 func (cp *copier) hostRoot(ctrRoot string) (string, error) {
 357         if ctrRoot == cp.ctrOutputDir {
 358                 return cp.hostOutputDir, nil
 359         }
 360         if mnt, ok := cp.bindmounts[ctrRoot]; ok {
 361                 return mnt.HostPath, nil
 362         }
 363         return "", fmt.Errorf("not bind-mounted: %q", ctrRoot)
 364 }
 365
 366 func (cp *copier) copyRegularFiles(m arvados.Mount) bool {
 367         return m.Kind == "text" || m.Kind == "json" || (m.Kind == "collection" && m.Writable)
 368 }
 369
 370 func (cp *copier) getManifest(pdh string) (*manifest.Manifest, error) {
 371         if mft, ok := cp.manifestCache[pdh]; ok {
 372                 return mft, nil
 373         }
 374         var coll arvados.Collection
 375         err := cp.arvClient.Get("collections", pdh, nil, &coll)
 376         if err != nil {
 377                 return nil, fmt.Errorf("error retrieving collection record for %q: %s", pdh, err)
 378         }
 379         mft := &manifest.Manifest{Text: coll.ManifestText}
 380         if cp.manifestCache == nil {
 381                 cp.manifestCache = map[string]*manifest.Manifest{pdh: mft}
 382         } else {
 383                 cp.manifestCache[pdh] = mft
 384         }
 385         return mft, nil
 386 }