isolate: pull uploading of deps out of exparchive main.

client/cmd/isolate/exp_archive.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ skipping 151 matching lines @@
         walker := partitioningWalker{fsView: fsView}
         for _, dep := range deps {
                 // Try to walk dep. If dep is a file (or symlink), the inner function is called exactly once.
                 if err := filepath.Walk(filepath.Clean(dep), walker.walkFn); err != nil {
                         return partitionedDeps{}, err
                 }
         }
         return walker.parts, nil
 }
 
-// main contains the core logic for experimental archive.
-func (c *expArchiveRun) main() error {
-        // TODO(djd): This func is long and has a lot of internal complexity (like,
-        // such as, archiveCallback). Refactor.
-
-        start := time.Now()
-        archiveOpts := &c.isolateFlags.ArchiveOptions
-        // Parse the incoming isolate file.
-        deps, rootDir, isol, err := isolate.ProcessIsolate(archiveOpts)
-        if err != nil {
-                return fmt.Errorf("failed to process isolate: %v", err)
-        }
-        log.Printf("Isolate referenced %d deps", len(deps))
-
-        // Set up a background context which is cancelled when this function returns.
-        ctx, cancel := context.WithCancel(context.Background())
-        defer cancel()
-
-        // Create the isolated client which connects to the isolate server.
-        authCl, err := c.createAuthClient()
-        if err != nil {
-                return err
-        }
-        client := isolatedclient.New(nil, authCl, c.isolatedFlags.ServerURL, c.isolatedFlags.Namespace, nil, nil)
-
-        // Set up a checker and uploader. We limit the uploader to one concurrent
-        // upload, since the uploads are all coming from disk (with the exception of
-        // the isolated JSON itself) and we only want a single goroutine reading from
-        // disk at once.
-        checker := NewChecker(ctx, client)
-        uploader := NewUploader(ctx, client, 1)
-
-        parts, err := partitionDeps(deps, rootDir, c.isolateFlags.ArchiveOptions.Blacklist)
-        if err != nil {
-                return fmt.Errorf("partitioning deps: %v", err)
-        }
-
+func uploadDeps(parts partitionedDeps, checker *Checker, uploader *Uploader) (map[string]isolated.File, error) {
         // Construct a map of the files that constitute the isolate.
         files := make(map[string]isolated.File)
 
-        numFiles := len(parts.filesToArchive.items) + len(parts.indivFiles.items)
-        filesSize := uint64(parts.filesToArchive.totalSize + parts.indivFiles.totalSize)
-        log.Printf("Isolate expanded to %d files (total size %s) and %d symlinks", numFiles, humanize.Bytes(filesSize), len(parts.links.items))
-        log.Printf("\t%d files (%s) to be isolated individually", len(parts.indivFiles.items), humanize.Bytes(uint64(parts.indivFiles.totalSize)))
-        log.Printf("\t%d files (%s) to be isolated in archives", len(parts.filesToArchive.items), humanize.Bytes(uint64(parts.filesToArchive.totalSize)))
-
         // Handle the symlinks.
         for _, item := range parts.links.items {
                 l, err := os.Readlink(item.Path)
                 if err != nil {
-                        return fmt.Errorf("unable to resolve symlink for %q: %v", item.Path, err)
+                        return nil, fmt.Errorf("unable to resolve symlink for %q: %v", item.Path, err)
                 }
                 files[item.RelPath] = isolated.SymLink(l)
         }
 
         // Handle the small to-be-archived files.
         bundles := ShardItems(parts.filesToArchive.items, archiveMaxSize)
         log.Printf("\t%d TAR archives to be isolated", len(bundles))
 
         for _, bundle := range bundles {
                 bundle := bundle
                 digest, tarSize, err := bundle.Digest()
                 if err != nil {
-                        return err
+                        return nil, err
                 }
 
                 log.Printf("Created tar archive %q (%s)", digest, humanize.Bytes(uint64(tarSize)))
                 log.Printf("\tcontains %d files (total %s)", len(bundle.Items), humanize.Bytes(uint64(bundle.ItemSize)))
                 // Mint an item for this tar.
                 item := &Item{
                         Path:    fmt.Sprintf(".%s.tar", digest),
                         RelPath: fmt.Sprintf(".%s.tar", digest),
                         Size:    tarSize,
                         Mode:    0644, // Read
                         Digest:  digest,
                 }
                 files[item.RelPath] = isolated.TarFile(item.Digest, int(item.Mode), item.Size)
 
                 checker.AddItem(item, false, func(item *Item, ps *isolatedclient.PushState) {
                         if ps == nil {
                                 return
                         }
                         log.Printf("QUEUED %q for upload", item.RelPath)
                         uploader.Upload(item.RelPath, bundle.Contents, ps, func() {
                                 log.Printf("UPLOADED %q", item.RelPath)
                         })
                 })
         }
 
         // Handle the large individually-uploaded files.
         for _, item := range parts.indivFiles.items {
                 d, err := hashFile(item.Path)
                 if err != nil {
-                        return err
+                        return nil, err
                 }
                 item.Digest = d
                 files[item.RelPath] = isolated.BasicFile(item.Digest, int(item.Mode), item.Size)
                 checker.AddItem(item, false, func(item *Item, ps *isolatedclient.PushState) {
                         if ps == nil {
                                 return
                         }
                         log.Printf("QUEUED %q for upload", item.RelPath)
                         uploader.UploadFile(item, ps, func() {
                                 log.Printf("UPLOADED %q", item.RelPath)
                         })
                 })
         }
+        return files, nil
+}
+
+// main contains the core logic for experimental archive.
+func (c *expArchiveRun) main() error {
+        // TODO(djd): This func is long and has a lot of internal complexity (like,
+        // such as, archiveCallback). Refactor.
+
+        start := time.Now()
+        archiveOpts := &c.isolateFlags.ArchiveOptions
+        // Parse the incoming isolate file.
+        deps, rootDir, isol, err := isolate.ProcessIsolate(archiveOpts)
+        if err != nil {
+                return fmt.Errorf("failed to process isolate: %v", err)
+        }
+        log.Printf("Isolate referenced %d deps", len(deps))
+
+        // Set up a background context which is cancelled when this function returns.
+        ctx, cancel := context.WithCancel(context.Background())
+        defer cancel()
+
+        // Create the isolated client which connects to the isolate server.
+        authCl, err := c.createAuthClient()
+        if err != nil {
+                return err
+        }
+        client := isolatedclient.New(nil, authCl, c.isolatedFlags.ServerURL, c.isolatedFlags.Namespace, nil, nil)
+
+        // Set up a checker and uploader. We limit the uploader to one concurrent
+        // upload, since the uploads are all coming from disk (with the exception of
+        // the isolated JSON itself) and we only want a single goroutine reading from
+        // disk at once.
+        checker := NewChecker(ctx, client)
+        uploader := NewUploader(ctx, client, 1)
+
+        parts, err := partitionDeps(deps, rootDir, c.isolateFlags.ArchiveOptions.Blacklist)
+        if err != nil {
+                return fmt.Errorf("partitioning deps: %v", err)
+        }
+
+        numFiles := len(parts.filesToArchive.items) + len(parts.indivFiles.items)
+        filesSize := uint64(parts.filesToArchive.totalSize + parts.indivFiles.totalSize)
+        log.Printf("Isolate expanded to %d files (total size %s) and %d symlinks", numFiles, humanize.Bytes(filesSize), len(parts.links.items))
+        log.Printf("\t%d files (%s) to be isolated individually", len(parts.indivFiles.items), humanize.Bytes(uint64(parts.indivFiles.totalSize)))
+        log.Printf("\t%d files (%s) to be isolated in archives", len(parts.filesToArchive.items), humanize.Bytes(uint64(parts.filesToArchive.totalSize)))
+
+        files, err := uploadDeps(parts, checker, uploader)
+        if err != nil {
+                return err
+        }
 
         // Marshal the isolated file into JSON, and create an Item to describe it.
         isol.Files = files
         var isolJSON []byte
         isolJSON, err = json.Marshal(isol)
         if err != nil {
                 return err
         }
         isolItem := &Item{
                 Path:    archiveOpts.Isolated,
@@ skipping 97 matching lines @@
 }
 
 func hashFile(path string) (isolated.HexDigest, error) {
         f, err := os.Open(path)
         if err != nil {
                 return "", err
         }
         defer f.Close()
         return isolated.Hash(f)
 }