X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/95ef2a6f4145d8c6058738b6de0bce098e3423d6..c22bd85568bdef0358a82bb20e4995477c7d24ac:/services/crunch-run/crunchrun.go

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index fd2ce3f659..0793729172 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -145,6 +145,7 @@ type ContainerRunner struct {
 	HostOutputDir  string
 	CleanupTempDir []string
 	Binds          []string
+	Volumes        map[string]struct{}
 	OutputPDH      *string
 	SigChan        chan os.Signal
 	ArvMountExit   chan error
@@ -336,6 +337,7 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 
 	collectionPaths := []string{}
 	runner.Binds = nil
+	runner.Volumes = make(map[string]struct{})
 	needCertMount := true
 
 	var binds []string
@@ -428,24 +430,25 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 			}
 			collectionPaths = append(collectionPaths, src)
 
-		case mnt.Kind == "tmp" && bind == runner.Container.OutputPath:
-			runner.HostOutputDir, err = runner.MkTempDir("", "")
+		case mnt.Kind == "tmp":
+			var tmpdir string
+			tmpdir, err = runner.MkTempDir("", "")
 			if err != nil {
 				return fmt.Errorf("While creating mount temp dir: %v", err)
 			}
-			st, staterr := os.Stat(runner.HostOutputDir)
+			st, staterr := os.Stat(tmpdir)
 			if staterr != nil {
 				return fmt.Errorf("While Stat on temp dir: %v", staterr)
 			}
-			err = os.Chmod(runner.HostOutputDir, st.Mode()|os.ModeSetgid|0777)
+			err = os.Chmod(tmpdir, st.Mode()|os.ModeSetgid|0777)
 			if staterr != nil {
 				return fmt.Errorf("While Chmod temp dir: %v", err)
 			}
-			runner.CleanupTempDir = append(runner.CleanupTempDir, runner.HostOutputDir)
-			runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", runner.HostOutputDir, bind))
-
-		case mnt.Kind == "tmp":
-			runner.Binds = append(runner.Binds, bind)
+			runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
+			runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", tmpdir, bind))
+			if bind == runner.Container.OutputPath {
+				runner.HostOutputDir = tmpdir
+			}
 
 		case mnt.Kind == "json":
 			jsondata, err := json.Marshal(mnt.Content)
@@ -631,11 +634,12 @@ func (runner *ContainerRunner) LogNodeInfo() (err error) {
 // Get and save the raw JSON container record from the API server
 func (runner *ContainerRunner) LogContainerRecord() (err error) {
 	w := &ArvLogWriter{
-		runner.ArvClient,
-		runner.Container.UUID,
-		"container",
-		runner.LogCollection.Open("container.json"),
+		ArvClient:     runner.ArvClient,
+		UUID:          runner.Container.UUID,
+		loggingStream: "container",
+		writeCloser:   runner.LogCollection.Open("container.json"),
 	}
+
 	// Get Container record JSON from the API Server
 	reader, err := runner.ArvClient.CallRaw("GET", "containers", runner.Container.UUID, "", nil)
 	if err != nil {
@@ -793,6 +797,8 @@ func (runner *ContainerRunner) CreateContainer() error {
 		runner.ContainerConfig.Env = append(runner.ContainerConfig.Env, k+"="+v)
 	}
 
+	runner.ContainerConfig.Volumes = runner.Volumes
+
 	runner.HostConfig = dockercontainer.HostConfig{
 		Binds:  runner.Binds,
 		Cgroup: dockercontainer.CgroupSpec(runner.setCgroupParent),
@@ -1061,8 +1067,8 @@ func (runner *ContainerRunner) CommitLogs() error {
 	// point, but re-open crunch log with ArvClient in case there are any
 	// other further (such as failing to write the log to Keep!) while
 	// shutting down
-	runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{runner.ArvClient, runner.Container.UUID,
-		"crunch-run", nil})
+	runner.CrunchLog = NewThrottledLogger(&ArvLogWriter{ArvClient: runner.ArvClient,
+		UUID: runner.Container.UUID, loggingStream: "crunch-run", writeCloser: nil})
 
 	if runner.LogsPDH != nil {
 		// If we have already assigned something to LogsPDH,
@@ -1149,7 +1155,8 @@ func (runner *ContainerRunner) IsCancelled() bool {
 
 // NewArvLogWriter creates an ArvLogWriter
 func (runner *ContainerRunner) NewArvLogWriter(name string) io.WriteCloser {
-	return &ArvLogWriter{runner.ArvClient, runner.Container.UUID, name, runner.LogCollection.Open(name + ".txt")}
+	return &ArvLogWriter{ArvClient: runner.ArvClient, UUID: runner.Container.UUID, loggingStream: name,
+		writeCloser: runner.LogCollection.Open(name + ".txt")}
 }
 
 // Run the full container lifecycle.
@@ -1289,6 +1296,9 @@ func NewContainerRunner(api IArvadosClient,
 	cr.Container.UUID = containerUUID
 	cr.CrunchLog = NewThrottledLogger(cr.NewLogWriter("crunch-run"))
 	cr.CrunchLog.Immediate = log.New(os.Stderr, containerUUID+" ", 0)
+
+	loadLogThrottleParams(api)
+
 	return cr
 }