[ARVADOS] updated: 9eafed4c378012e914000658b30af4f8a1c51432
git at public.curoverse.com
git at public.curoverse.com
Fri Feb 12 20:22:44 EST 2016
Summary of changes:
services/crunch-run/crunchrun.go | 171 ++++++++++++++++++++++++++--------
services/crunch-run/crunchrun_test.go | 4 +-
services/crunch-run/upload.go | 52 +++++------
3 files changed, 160 insertions(+), 67 deletions(-)
via 9eafed4c378012e914000658b30af4f8a1c51432 (commit)
via b47e664b17efbedd41b868d05164272d5549ffc1 (commit)
from c8cba416487f240c6e8395b1581d9f4e441cc5f7 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit 9eafed4c378012e914000658b30af4f8a1c51432
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Fri Feb 12 17:49:28 2016 -0500
8015: Fix compilation
diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index a7d8a58..5f63f5d 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -1,14 +1,16 @@
package main
import (
+ "encoding/json"
"errors"
"flag"
+ "fmt"
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
"git.curoverse.com/arvados.git/sdk/go/keepclient"
"git.curoverse.com/arvados.git/sdk/go/manifest"
"github.com/curoverse/dockerclient"
"io"
- "ioutil"
+ "io/ioutil"
"log"
"os"
"os/exec"
@@ -178,7 +180,11 @@ func (runner *ContainerRunner) LoadImage() (err error) {
}
func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig) (err error) {
- runner.ArvMountPoint = ioutil.TempDir("", "keep")
+ runner.ArvMountPoint, err = ioutil.TempDir("", "keep")
+ if err != nil {
+ return fmt.Errorf("While creating keep mount temp dir: %v", err)
+ }
+
pdhOnly := true
tmpcount := 0
arvMountCmd := []string{"--foreground"}
@@ -195,14 +201,14 @@ func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig)
return fmt.Errorf("Writing to existing collections currently not permitted.")
}
pdhOnly = false
- src = fmt.Sprintf("%s/by_id/%s", arvMountPoint, mnt.UUID)
+ src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.UUID)
} else if mnt.PortableDataHash != "" {
if mnt.Writable {
return fmt.Errorf("Can never write to a collection specified by portable data hash")
}
- src = fmt.Sprintf("%s/by_id/%s", arvMountPoint, mnt.PortableDataHash)
+ src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
} else {
- src = fmt.Sprintf("%s/tmp%i", arvMountPoint, tmpcount)
+ src = fmt.Sprintf("%s/tmp%i", runner.ArvMountPoint, tmpcount)
arvMountCmd = append(arvMountCmd, "--mount-tmp")
arvMountCmd = append(arvMountCmd, fmt.Sprintf("tmp%i", tmpcount))
tmpcount += 1
@@ -215,10 +221,14 @@ func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig)
} else {
hostConfig.Binds = append(hostConfig.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
}
- collectionPaths = append(collections, src)
+ collectionPaths = append(collectionPaths, src)
} else if mnt.Kind == "tmp" {
if bind == runner.ContainerRecord.OutputPath {
- runner.HostOutputDir = ioutil.TempDir("", "")
+ runner.HostOutputDir, err = ioutil.TempDir("", "")
+ if err != nil {
+ return fmt.Errorf("While creating mount temp dir: %v", err)
+ }
+
hostConfig.Binds = append(hostConfig.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
} else {
hostConfig.Binds = append(hostConfig.Binds, bind)
@@ -237,16 +247,16 @@ func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig)
} else {
arvMountCmd = append(arvMountCmd, "--mount-by-id", "by_id")
}
- arvMountCmd = append(arvMountCmd, arvMountPoint)
+ arvMountCmd = append(arvMountCmd, runner.ArvMountPoint)
- runner.ArvMount = exec.Command("arv-mount", arvMountCmd)
+ runner.ArvMount = exec.Command("arv-mount", arvMountCmd...)
err = runner.ArvMount.Start()
if err != nil {
runner.ArvMount = nil
return fmt.Errorf("While trying to start arv-mount: %v", err)
}
- for p := range collectionPaths {
+ for _, p := range collectionPaths {
_, err = os.Stat(p)
if err != nil {
return fmt.Errorf("While checking that input files exist: %v", err)
@@ -353,19 +363,19 @@ func (runner *ContainerRunner) HandleOutput() error {
return nil
}
- _, err = os.Stat(os.HostOutputPath)
+ _, err := os.Stat(runner.HostOutputDir)
if err != nil {
return fmt.Errorf("While checking host output path: %v", err)
}
var manifestText string
- collectionMetafile = fmt.Sprintf("%s/.arvados#collection", os.HostOutputPath)
+ collectionMetafile := fmt.Sprintf("%s/.arvados#collection", runner.HostOutputDir)
_, err = os.Stat(collectionMetafile)
if err != nil {
// Regular directory
- cw := CollectionWriter{runner.Kc}
- manifestText, err = cw.WriteTree(os.HostOutputPath, runner.CrunchLog)
+ cw := CollectionWriter{runner.Kc, nil, sync.Mutex{}}
+ manifestText, err = cw.WriteTree(runner.HostOutputDir, runner.CrunchLog.Logger)
if err != nil {
return fmt.Errorf("While uploading output files: %v", err)
}
@@ -386,7 +396,9 @@ func (runner *ContainerRunner) HandleOutput() error {
}
var response CollectionRecord
- err = runner.ArvClient.Create("collections", &response)
+ err = runner.ArvClient.Create("collections",
+ arvadosclient.Dict{"manifest_text": manifestText},
+ &response)
if err != nil {
return fmt.Errorf("While creating output collection: %v", err)
}
@@ -476,7 +488,7 @@ func (runner *ContainerRunner) Run(containerUUID string) (err error) {
// (6) handle output
outputerr := runner.HandleOutput()
if outputerr != nil {
- runner.CrunchLog.Print(outputrr)
+ runner.CrunchLog.Print(outputerr)
}
// (7) write logs
@@ -557,7 +569,7 @@ func NewContainerRunner(api IArvadosClient,
cr := &ContainerRunner{ArvClient: api, Kc: kc, Docker: docker}
cr.NewLogWriter = cr.NewArvLogWriter
- cr.LogCollection = &CollectionWriter{kc, nil}
+ cr.LogCollection = &CollectionWriter{kc, nil, sync.Mutex{}}
cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
return cr
}
diff --git a/services/crunch-run/crunchrun_test.go b/services/crunch-run/crunchrun_test.go
index 1946e5c..3c4dbf0 100644
--- a/services/crunch-run/crunchrun_test.go
+++ b/services/crunch-run/crunchrun_test.go
@@ -161,9 +161,9 @@ func (this *ArvTestClient) Create(resourceType string,
func (this *ArvTestClient) Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error {
if resourceType == "collections" {
if uuid == hwPDH {
- output.(*Collection).ManifestText = hwManifest
+ output.(*CollectionRecord).ManifestText = hwManifest
} else if uuid == otherPDH {
- output.(*Collection).ManifestText = otherManifest
+ output.(*CollectionRecord).ManifestText = otherManifest
}
}
if resourceType == "containers" {
diff --git a/services/crunch-run/upload.go b/services/crunch-run/upload.go
index 9a62e2e..3285fcc 100644
--- a/services/crunch-run/upload.go
+++ b/services/crunch-run/upload.go
@@ -18,6 +18,8 @@ import (
"git.curoverse.com/arvados.git/sdk/go/manifest"
"io"
"log"
+ "os"
+ "path/filepath"
"strings"
"sync"
)
@@ -208,22 +210,24 @@ func (m *CollectionWriter) ManifestText() (mt string, err error) {
return buf.String(), nil
}
+type WalkUpload struct {
+ kc IKeepClient
+ stripPrefix string
+ streamMap map[string]*CollectionFileWriter
+ status *log.Logger
+}
+
// WalkFunc walks a directory tree, uploads each file found and adds it to the
// CollectionWriter.
-func (m *CollectionWriter) WalkFunc(path string,
- info os.FileInfo,
- err error,
- stripPrefix string,
- streamMap map[string]*manifest.ManifestStream,
- status log.Logger) error {
+func (m *WalkUpload) WalkFunc(path string, info os.FileInfo, err error) error {
if info.IsDir() {
return nil
}
var dir string
- if len(path) > (len(stripPrefix) + len(info.Name()) + 1) {
- dir = path[len(stripPrefix)+1 : (len(path) - len(info.Name()) - 1)]
+ if len(path) > (len(m.stripPrefix) + len(info.Name()) + 1) {
+ dir = path[len(m.stripPrefix)+1 : (len(path) - len(info.Name()) - 1)]
}
if dir == "" {
dir = "."
@@ -231,9 +235,9 @@ func (m *CollectionWriter) WalkFunc(path string,
fn := path[(len(path) - len(info.Name())):]
- if streamMap[dir] == nil {
- streamMap[dir] = &CollectionFileWriter{
- m.IKeepClient,
+ if m.streamMap[dir] == nil {
+ m.streamMap[dir] = &CollectionFileWriter{
+ m.kc,
&manifest.ManifestStream{StreamName: dir},
0,
0,
@@ -241,10 +245,10 @@ func (m *CollectionWriter) WalkFunc(path string,
make(chan *Block),
make(chan []error),
""}
- go streamMap[dir].goUpload()
+ go m.streamMap[dir].goUpload()
}
- fileWriter := streamMap[dir]
+ fileWriter := m.streamMap[dir]
// Reset the CollectionFileWriter for a new file
fileWriter.NewFile(fn)
@@ -255,10 +259,9 @@ func (m *CollectionWriter) WalkFunc(path string,
}
defer file.Close()
- status.Printf("Uploading %v/%v (%v bytes)", dir, fn, info.Size())
+ m.status.Printf("Uploading %v/%v (%v bytes)", dir, fn, info.Size())
- var count int64
- count, err = io.Copy(fileWriter, file)
+ _, err = io.Copy(fileWriter, file)
if err != nil {
return err
}
@@ -269,16 +272,10 @@ func (m *CollectionWriter) WalkFunc(path string,
return nil
}
-func (cw *CollectionWriter) WriteTree(root string, status log.Logger) (manifest string, err error) {
- streamMap := make(map[string]*ManifestStreamWriter)
- err = filepath.Walk(root, func(path string, info os.FileInfo, err error) {
- return cw.WalkFunc(path,
- info,
- err,
- root,
- streamMap,
- status)
- })
+func (cw *CollectionWriter) WriteTree(root string, status *log.Logger) (manifest string, err error) {
+ streamMap := make(map[string]*CollectionFileWriter)
+ wu := &WalkUpload{cw.IKeepClient, root, streamMap, status}
+ err = filepath.Walk(root, wu.WalkFunc)
if err != nil {
return "", err
@@ -290,5 +287,5 @@ func (cw *CollectionWriter) WriteTree(root string, status log.Logger) (manifest
}
cw.mtx.Unlock()
- return mw.ManifestText()
+ return cw.ManifestText()
}
commit b47e664b17efbedd41b868d05164272d5549ffc1
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Fri Feb 12 17:38:58 2016 -0500
8015: Finished initial implemention of input and output mounts. Needs tests.
diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 8f6f05c..a7d8a58 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -44,8 +44,9 @@ type Mount struct {
}
// Collection record returned by the API server.
-type Collection struct {
- ManifestText string `json:"manifest_text"`
+type CollectionRecord struct {
+ ManifestText string `json:"manifest_text"`
+ PortableDataHash string `json:"portable_data_hash"`
}
// ContainerRecord is the container record returned by the API server.
@@ -60,6 +61,7 @@ type ContainerRecord struct {
Priority int `json:"priority"`
RuntimeConstraints map[string]interface{} `json:"runtime_constraints"`
State string `json:"state"`
+ Output string `json:"output"`
}
// NewLogWriter is a factory function to create a new log writer.
@@ -96,6 +98,8 @@ type ContainerRunner struct {
LogsPDH *string
ArvMount *exec.Cmd
ArvMountPoint string
+ HostOutputDir string
+ OutputPDH string
CancelLock sync.Mutex
Cancelled bool
SigChan chan os.Signal
@@ -133,17 +137,17 @@ func (runner *ContainerRunner) LoadImage() (err error) {
runner.CrunchLog.Printf("Fetching Docker image from collection '%s'", runner.ContainerRecord.ContainerImage)
- var collection Collection
+ var collection CollectionRecord
err = runner.ArvClient.Get("collections", runner.ContainerRecord.ContainerImage, nil, &collection)
if err != nil {
- return err
+ return fmt.Errorf("While getting container image collection: %v", err)
}
manifest := manifest.Manifest{Text: collection.ManifestText}
var img, imageID string
for ms := range manifest.StreamIter() {
img = ms.FileStreamSegments[0].Name
if !strings.HasSuffix(img, ".tar") {
- return errors.New("First file in the collection does not end in .tar")
+ return fmt.Errorf("First file in the container image collection does not end in .tar")
}
imageID = img[:len(img)-4]
}
@@ -157,12 +161,12 @@ func (runner *ContainerRunner) LoadImage() (err error) {
var readCloser io.ReadCloser
readCloser, err = runner.Kc.ManifestFileReader(manifest, img)
if err != nil {
- return err
+ return fmt.Errorf("While creating ManifestFileReader for container image: %v", err)
}
err = runner.Docker.LoadImage(readCloser)
if err != nil {
- return err
+ return fmt.Errorf("While loading container image into Docker: %v", err)
}
} else {
runner.CrunchLog.Print("Docker image is available")
@@ -178,17 +182,17 @@ func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig)
pdhOnly := true
tmpcount := 0
arvMountCmd := []string{"--foreground"}
- collections := []string{}
+ collectionPaths := []string{}
for bind, mnt := range runner.ContainerRecord.Mounts {
if mnt.Kind == "collection" {
var src string
if mnt.UUID != "" && mnt.PortableDataHash != "" {
- return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection")
+ return fmt.Errorf("Cannot specify both 'uuid' and 'portable_data_hash' for a collection mount")
}
if mnt.UUID != "" {
if mnt.Writable {
- return fmt.Errorf("Writing to collection currently not permitted.")
+ return fmt.Errorf("Writing to existing collections currently not permitted.")
}
pdhOnly = false
src = fmt.Sprintf("%s/by_id/%s", arvMountPoint, mnt.UUID)
@@ -204,18 +208,30 @@ func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig)
tmpcount += 1
}
if mnt.Writable {
+ if bind == runner.ContainerRecord.OutputPath {
+ runner.HostOutputDir = src
+ }
hostConfig.Binds = append(hostConfig.Binds, fmt.Sprintf("%s:%s", src, bind))
} else {
hostConfig.Binds = append(hostConfig.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
}
- collections = append(collections, src)
+ collectionPaths = append(collections, src)
} else if mnt.Kind == "tmp" {
- hostConfig.Binds = append(hostConfig.Binds, bind)
+ if bind == runner.ContainerRecord.OutputPath {
+ runner.HostOutputDir = ioutil.TempDir("", "")
+ hostConfig.Binds = append(hostConfig.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
+ } else {
+ hostConfig.Binds = append(hostConfig.Binds, bind)
+ }
} else {
return fmt.Errorf("Unknown mount kind '%s'", mnt.Kind)
}
}
+ if runner.HostOutputDir == "" {
+ return fmt.Errorf("Output path does not correspond to a writable mount point")
+ }
+
if pdhOnly {
arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
} else {
@@ -226,11 +242,18 @@ func (runner *ContainerRunner) SetupMounts(hostConfig *dockerclient.HostConfig)
runner.ArvMount = exec.Command("arv-mount", arvMountCmd)
err = runner.ArvMount.Start()
if err != nil {
- return err
+ runner.ArvMount = nil
+ return fmt.Errorf("While trying to start arv-mount: %v", err)
}
- // XXX need to go through and os.Stat() each file or dir in "sources"
- // to make sure they show up for Docker.
+ for p := range collectionPaths {
+ _, err = os.Stat(p)
+ if err != nil {
+ return fmt.Errorf("While checking that input files exist: %v", err)
+ }
+ }
+
+ return nil
}
// StartContainer creates the container and runs it.
@@ -253,19 +276,19 @@ func (runner *ContainerRunner) StartContainer() (err error) {
}
runner.ContainerID, err = runner.Docker.CreateContainer(&runner.ContainerConfig, "", nil)
if err != nil {
- return
+ return fmt.Errorf("While creating container: %v", err)
}
hostConfig := &dockerclient.HostConfig{}
err = runner.SetupMounts(hostConfig)
if err != nil {
- return
+ return fmt.Errorf("While setting up mounts: %v", err)
}
runner.CrunchLog.Printf("Starting Docker container id '%s'", runner.ContainerID)
err = runner.Docker.StartContainer(runner.ContainerID, hostConfig)
if err != nil {
- return
+ return fmt.Errorf("While starting container: %v", err)
}
return nil
@@ -280,11 +303,11 @@ func (runner *ContainerRunner) AttachLogs() (err error) {
var stderrReader, stdoutReader io.Reader
stderrReader, err = runner.Docker.ContainerLogs(runner.ContainerID, &dockerclient.LogOptions{Follow: true, Stderr: true})
if err != nil {
- return
+ return fmt.Errorf("While getting container standard error: %v", err)
}
stdoutReader, err = runner.Docker.ContainerLogs(runner.ContainerID, &dockerclient.LogOptions{Follow: true, Stdout: true})
if err != nil {
- return
+ return fmt.Errorf("While getting container standard output: %v", err)
}
runner.loggingDone = make(chan bool)
@@ -303,7 +326,7 @@ func (runner *ContainerRunner) WaitFinish() error {
result := runner.Docker.Wait(runner.ContainerID)
wr := <-result
if wr.Error != nil {
- return wr.Error
+ return fmt.Errorf("While waiting for container to finish: %v", wr.Error)
}
runner.ExitCode = &wr.ExitCode
@@ -314,8 +337,61 @@ func (runner *ContainerRunner) WaitFinish() error {
runner.Stdout.Close()
runner.Stderr.Close()
- umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
- umount.Run()
+ return nil
+}
+
+// HandleOutput sets the output and unmounts the FUSE mount.
+func (runner *ContainerRunner) HandleOutput() error {
+ if runner.ArvMount != nil {
+ defer func() {
+ umount := exec.Command("fusermount", "-z", "-u", runner.ArvMountPoint)
+ umount.Run()
+ }()
+ }
+
+ if runner.finalState != "Complete" {
+ return nil
+ }
+
+ _, err = os.Stat(os.HostOutputPath)
+ if err != nil {
+ return fmt.Errorf("While checking host output path: %v", err)
+ }
+
+ var manifestText string
+
+ collectionMetafile = fmt.Sprintf("%s/.arvados#collection", os.HostOutputPath)
+ _, err = os.Stat(collectionMetafile)
+ if err != nil {
+ // Regular directory
+ cw := CollectionWriter{runner.Kc}
+ manifestText, err = cw.WriteTree(os.HostOutputPath, runner.CrunchLog)
+ if err != nil {
+ return fmt.Errorf("While uploading output files: %v", err)
+ }
+ } else {
+ // FUSE mount directory
+ file, openerr := os.Open(collectionMetafile)
+ if openerr != nil {
+ return fmt.Errorf("While opening FUSE metafile: %v", err)
+ }
+ defer file.Close()
+
+ rec := CollectionRecord{}
+ err = json.NewDecoder(file).Decode(&rec)
+ if err != nil {
+ return fmt.Errorf("While reading FUSE metafile: %v", err)
+ }
+ manifestText = rec.ManifestText
+ }
+
+ var response CollectionRecord
+ err = runner.ArvClient.Create("collections", &response)
+ if err != nil {
+ return fmt.Errorf("While creating output collection: %v", err)
+ }
+
+ runner.OutputPDH = response.PortableDataHash
return nil
}
@@ -334,7 +410,7 @@ func (runner *ContainerRunner) CommitLogs() error {
mt, err := runner.LogCollection.ManifestText()
if err != nil {
- return err
+ return fmt.Errorf("While creating log manifest: %v", err)
}
response := make(map[string]string)
@@ -343,7 +419,7 @@ func (runner *ContainerRunner) CommitLogs() error {
"manifest_text": mt},
response)
if err != nil {
- return err
+ return fmt.Errorf("While creating log collection: %v", err)
}
runner.LogsPDH = new(string)
@@ -370,6 +446,7 @@ func (runner *ContainerRunner) UpdateContainerRecordComplete() error {
}
update["state"] = runner.finalState
+ update["output"] = runner.OutputPDH
return runner.ArvClient.Update("containers", runner.ContainerRecord.UUID, update, nil)
}
@@ -396,13 +473,19 @@ func (runner *ContainerRunner) Run(containerUUID string) (err error) {
runner.finalState = "Complete"
}
- // (6) write logs
+ // (6) handle output
+ outputerr := runner.HandleOutput()
+ if outputerr != nil {
+ runner.CrunchLog.Print(outputrr)
+ }
+
+ // (7) write logs
logerr := runner.CommitLogs()
if logerr != nil {
runner.CrunchLog.Print(logerr)
}
- // (7) update container record with results
+ // (8) update container record with results
updateerr := runner.UpdateContainerRecordComplete()
if updateerr != nil {
runner.CrunchLog.Print(updateerr)
@@ -425,19 +508,19 @@ func (runner *ContainerRunner) Run(containerUUID string) (err error) {
err = runner.ArvClient.Get("containers", containerUUID, nil, &runner.ContainerRecord)
if err != nil {
- return
+ return fmt.Errorf("While getting container record: %v", err)
}
// (0) setup signal handling
err = runner.SetupSignals()
if err != nil {
- return
+ return fmt.Errorf("While setting up signal handling: %v", err)
}
// (1) check for and/or load image
err = runner.LoadImage()
if err != nil {
- return
+ return fmt.Errorf("While loading container image: %v", err)
}
// (2) start container
diff --git a/services/crunch-run/upload.go b/services/crunch-run/upload.go
index c3b8c37..9a62e2e 100644
--- a/services/crunch-run/upload.go
+++ b/services/crunch-run/upload.go
@@ -253,6 +253,7 @@ func (m *CollectionWriter) WalkFunc(path string,
if err != nil {
return err
}
+ defer file.Close()
status.Printf("Uploading %v/%v (%v bytes)", dir, fn, info.Size())
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list