X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b7b9ea44ada30b1251fb10c872cb1da1d7c29bd0..8ed521f7fd1e48e1e415125745ed8c6627a62c91:/services/crunch-run/crunchrun.go?ds=sidebyside

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 32d524abca..0b59f7df91 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -18,6 +18,8 @@ import (
 	"os/exec"
 	"os/signal"
 	"path"
+	"path/filepath"
+	"sort"
 	"strings"
 	"sync"
 	"syscall"
@@ -28,8 +30,9 @@ import (
 type IArvadosClient interface {
 	Create(resourceType string, parameters arvadosclient.Dict, output interface{}) error
 	Get(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
-	Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) (err error)
-	Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) (err error)
+	Update(resourceType string, uuid string, parameters arvadosclient.Dict, output interface{}) error
+	Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error
+	Discovery(key string) (interface{}, error)
 }
 
 // ErrCancelled is the error returned when the container is cancelled.
@@ -38,7 +41,7 @@ var ErrCancelled = errors.New("Cancelled")
 // IKeepClient is the minimal Keep API methods used by crunch-run.
 type IKeepClient interface {
 	PutHB(hash string, buf []byte) (string, int, error)
-	ManifestFileReader(m manifest.Manifest, filename string) (keepclient.ReadCloserWithLen, error)
+	ManifestFileReader(m manifest.Manifest, filename string) (keepclient.Reader, error)
 }
 
 // NewLogWriter is a factory function to create a new log writer.
@@ -97,7 +100,19 @@ type ContainerRunner struct {
 	statReporter *crunchstat.Reporter
 	statInterval time.Duration
 	cgroupRoot   string
-	cgroupParent string
+	// What we expect the container's cgroup parent to be.
+	expectCgroupParent string
+	// What we tell docker to use as the container's cgroup
+	// parent. Note: Ideally we would use the same field for both
+	// expectCgroupParent and setCgroupParent, and just make it
+	// default to "docker". However, when using docker < 1.10 with
+	// systemd, specifying a non-empty cgroup parent (even the
+	// default value "docker") hits a docker bug
+	// (https://github.com/docker/docker/issues/17126). Using two
+	// separate fields makes it possible to use the "expect cgroup
+	// parent to be X" feature even on sites where the "specify
+	// cgroup parent" feature breaks.
+	setCgroupParent string
 }
 
 // SetupSignals sets up signal handling to gracefully terminate the underlying
@@ -224,8 +239,15 @@ func (runner *ContainerRunner) ArvMountCmd(arvMountCmd []string, token string) (
 	return c, nil
 }
 
+func (runner *ContainerRunner) SetupArvMountPoint(prefix string) (err error) {
+	if runner.ArvMountPoint == "" {
+		runner.ArvMountPoint, err = runner.MkTempDir("", prefix)
+	}
+	return
+}
+
 func (runner *ContainerRunner) SetupMounts() (err error) {
-	runner.ArvMountPoint, err = runner.MkTempDir("", "keep")
+	err = runner.SetupArvMountPoint("keep")
 	if err != nil {
 		return fmt.Errorf("While creating keep mount temp dir: %v", err)
 	}
@@ -235,10 +257,23 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 	pdhOnly := true
 	tmpcount := 0
 	arvMountCmd := []string{"--foreground", "--allow-other", "--read-write"}
+
+	if runner.Container.RuntimeConstraints.KeepCacheRAM > 0 {
+		arvMountCmd = append(arvMountCmd, "--file-cache", fmt.Sprintf("%d", runner.Container.RuntimeConstraints.KeepCacheRAM))
+	}
+
 	collectionPaths := []string{}
 	runner.Binds = nil
+	needCertMount := true
 
-	for bind, mnt := range runner.Container.Mounts {
+	var binds []string
+	for bind, _ := range runner.Container.Mounts {
+		binds = append(binds, bind)
+	}
+	sort.Strings(binds)
+
+	for _, bind := range binds {
+		mnt := runner.Container.Mounts[bind]
 		if bind == "stdout" {
 			// Is it a "file" mount kind?
 			if mnt.Kind != "file" {
@@ -255,7 +290,18 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 			}
 		}
 
-		if mnt.Kind == "collection" {
+		if bind == "/etc/arvados/ca-certificates.crt" {
+			needCertMount = false
+		}
+
+		if strings.HasPrefix(bind, runner.Container.OutputPath+"/") && bind != runner.Container.OutputPath+"/" {
+			if mnt.Kind != "collection" {
+				return fmt.Errorf("Only mount points of kind 'collection' are supported underneath the output_path: %v", bind)
+			}
+		}
+
+		switch {
+		case 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 mount")
@@ -270,7 +316,21 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 				if mnt.Writable {
 					return fmt.Errorf("Can never write to a collection specified by portable data hash")
 				}
+				idx := strings.Index(mnt.PortableDataHash, "/")
+				if idx > 0 {
+					mnt.Path = path.Clean(mnt.PortableDataHash[idx:])
+					mnt.PortableDataHash = mnt.PortableDataHash[0:idx]
+					runner.Container.Mounts[bind] = mnt
+				}
 				src = fmt.Sprintf("%s/by_id/%s", runner.ArvMountPoint, mnt.PortableDataHash)
+				if mnt.Path != "" && mnt.Path != "." {
+					if strings.HasPrefix(mnt.Path, "./") {
+						mnt.Path = mnt.Path[2:]
+					} else if strings.HasPrefix(mnt.Path, "/") {
+						mnt.Path = mnt.Path[1:]
+					}
+					src += "/" + mnt.Path
+				}
 			} else {
 				src = fmt.Sprintf("%s/tmp%d", runner.ArvMountPoint, tmpcount)
 				arvMountCmd = append(arvMountCmd, "--mount-tmp")
@@ -280,31 +340,55 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 			if mnt.Writable {
 				if bind == runner.Container.OutputPath {
 					runner.HostOutputDir = src
+				} else if strings.HasPrefix(bind, runner.Container.OutputPath+"/") {
+					return fmt.Errorf("Writable mount points are not permitted underneath the output_path: %v", bind)
 				}
 				runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s", src, bind))
 			} else {
 				runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", src, bind))
 			}
 			collectionPaths = append(collectionPaths, src)
-		} else if mnt.Kind == "tmp" {
-			if bind == runner.Container.OutputPath {
-				runner.HostOutputDir, err = runner.MkTempDir("", "")
-				if err != nil {
-					return fmt.Errorf("While creating mount temp dir: %v", err)
-				}
-				st, staterr := os.Stat(runner.HostOutputDir)
-				if staterr != nil {
-					return fmt.Errorf("While Stat on temp dir: %v", staterr)
-				}
-				err = os.Chmod(runner.HostOutputDir, 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))
-			} else {
-				runner.Binds = append(runner.Binds, bind)
+
+		case mnt.Kind == "tmp" && bind == runner.Container.OutputPath:
+			runner.HostOutputDir, err = runner.MkTempDir("", "")
+			if err != nil {
+				return fmt.Errorf("While creating mount temp dir: %v", err)
+			}
+			st, staterr := os.Stat(runner.HostOutputDir)
+			if staterr != nil {
+				return fmt.Errorf("While Stat on temp dir: %v", staterr)
+			}
+			err = os.Chmod(runner.HostOutputDir, 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)
+
+		case mnt.Kind == "json":
+			jsondata, err := json.Marshal(mnt.Content)
+			if err != nil {
+				return fmt.Errorf("encoding json data: %v", err)
+			}
+			// Create a tempdir with a single file
+			// (instead of just a tempfile): this way we
+			// can ensure the file is world-readable
+			// inside the container, without having to
+			// make it world-readable on the docker host.
+			tmpdir, err := runner.MkTempDir("", "")
+			if err != nil {
+				return fmt.Errorf("creating temp dir: %v", err)
 			}
+			runner.CleanupTempDir = append(runner.CleanupTempDir, tmpdir)
+			tmpfn := filepath.Join(tmpdir, "mountdata.json")
+			err = ioutil.WriteFile(tmpfn, jsondata, 0644)
+			if err != nil {
+				return fmt.Errorf("writing temp file: %v", err)
+			}
+			runner.Binds = append(runner.Binds, fmt.Sprintf("%s:%s:ro", tmpfn, bind))
 		}
 	}
 
@@ -312,6 +396,16 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 		return fmt.Errorf("Output path does not correspond to a writable mount point")
 	}
 
+	if wantAPI := runner.Container.RuntimeConstraints.API; needCertMount && wantAPI != nil && *wantAPI {
+		for _, certfile := range arvadosclient.CertFiles {
+			_, err := os.Stat(certfile)
+			if err == nil {
+				runner.Binds = append(runner.Binds, fmt.Sprintf("%s:/etc/arvados/ca-certificates.crt:ro", certfile))
+				break
+			}
+		}
+	}
+
 	if pdhOnly {
 		arvMountCmd = append(arvMountCmd, "--mount-by-pdh", "by_id")
 	} else {
@@ -393,7 +487,7 @@ func (runner *ContainerRunner) StartCrunchstat() {
 	runner.statReporter = &crunchstat.Reporter{
 		CID:          runner.ContainerID,
 		Logger:       log.New(runner.statLogger, "", 0),
-		CgroupParent: runner.cgroupParent,
+		CgroupParent: runner.expectCgroupParent,
 		CgroupRoot:   runner.cgroupRoot,
 		PollPeriod:   runner.statInterval,
 	}
@@ -480,8 +574,13 @@ func (runner *ContainerRunner) CreateContainer() error {
 		return fmt.Errorf("While creating container: %v", err)
 	}
 
-	runner.HostConfig = dockerclient.HostConfig{Binds: runner.Binds,
-		LogConfig: dockerclient.LogConfig{Type: "none"}}
+	runner.HostConfig = dockerclient.HostConfig{
+		Binds:        runner.Binds,
+		CgroupParent: runner.setCgroupParent,
+		LogConfig: dockerclient.LogConfig{
+			Type: "none",
+		},
+	}
 
 	return runner.AttachStreams()
 }
@@ -520,6 +619,21 @@ func (runner *ContainerRunner) CaptureOutput() error {
 		return nil
 	}
 
+	if wantAPI := runner.Container.RuntimeConstraints.API; wantAPI != nil && *wantAPI {
+		// Output may have been set directly by the container, so
+		// refresh the container record to check.
+		err := runner.ArvClient.Get("containers", runner.Container.UUID,
+			nil, &runner.Container)
+		if err != nil {
+			return err
+		}
+		if runner.Container.Output != "" {
+			// Container output is already set.
+			runner.OutputPDH = &runner.Container.Output
+			return nil
+		}
+	}
+
 	if runner.HostOutputDir == "" {
 		return nil
 	}
@@ -556,20 +670,93 @@ func (runner *ContainerRunner) CaptureOutput() error {
 		manifestText = rec.ManifestText
 	}
 
+	// Pre-populate output from the configured mount points
+	var binds []string
+	for bind, _ := range runner.Container.Mounts {
+		binds = append(binds, bind)
+	}
+	sort.Strings(binds)
+
+	for _, bind := range binds {
+		mnt := runner.Container.Mounts[bind]
+
+		bindSuffix := strings.TrimPrefix(bind, runner.Container.OutputPath)
+
+		if bindSuffix == bind || len(bindSuffix) <= 0 {
+			// either does not start with OutputPath or is OutputPath itself
+			continue
+		}
+
+		if mnt.ExcludeFromOutput == true {
+			continue
+		}
+
+		// append to manifest_text
+		m, err := runner.getCollectionManifestForPath(mnt, bindSuffix)
+		if err != nil {
+			return err
+		}
+
+		manifestText = manifestText + m
+	}
+
+	// Save output
 	var response arvados.Collection
+	manifest := manifest.Manifest{Text: manifestText}
+	manifestText = manifest.Extract(".", ".").Text
 	err = runner.ArvClient.Create("collections",
 		arvadosclient.Dict{
 			"collection": arvadosclient.Dict{
+				"is_trashed":    true,
+				"name":          "output for " + runner.Container.UUID,
 				"manifest_text": manifestText}},
 		&response)
 	if err != nil {
 		return fmt.Errorf("While creating output collection: %v", err)
 	}
+	runner.OutputPDH = &response.PortableDataHash
+	return nil
+}
 
-	runner.OutputPDH = new(string)
-	*runner.OutputPDH = response.PortableDataHash
+var outputCollections = make(map[string]arvados.Collection)
+
+// Fetch the collection for the mnt.PortableDataHash
+// Return the manifest_text fragment corresponding to the specified mnt.Path
+//  after making any required updates.
+//  Ex:
+//    If mnt.Path is not specified,
+//      return the entire manifest_text after replacing any "." with bindSuffix
+//    If mnt.Path corresponds to one stream,
+//      return the manifest_text for that stream after replacing that stream name with bindSuffix
+//    Otherwise, check if a filename in any one stream is being sought. Return the manifest_text
+//      for that stream after replacing stream name with bindSuffix minus the last word
+//      and the file name with last word of the bindSuffix
+//  Allowed path examples:
+//    "path":"/"
+//    "path":"/subdir1"
+//    "path":"/subdir1/subdir2"
+//    "path":"/subdir/filename" etc
+func (runner *ContainerRunner) getCollectionManifestForPath(mnt arvados.Mount, bindSuffix string) (string, error) {
+	collection := outputCollections[mnt.PortableDataHash]
+	if collection.PortableDataHash == "" {
+		err := runner.ArvClient.Get("collections", mnt.PortableDataHash, nil, &collection)
+		if err != nil {
+			return "", fmt.Errorf("While getting collection for %v: %v", mnt.PortableDataHash, err)
+		}
+		outputCollections[mnt.PortableDataHash] = collection
+	}
 
-	return nil
+	if collection.ManifestText == "" {
+		runner.CrunchLog.Printf("No manifest text for collection %v", collection.PortableDataHash)
+		return "", nil
+	}
+
+	mft := manifest.Manifest{Text: collection.ManifestText}
+	extracted := mft.Extract(mnt.Path, bindSuffix)
+	if extracted.Err != nil {
+		return "", fmt.Errorf("Error parsing manifest for %v: %v", mnt.PortableDataHash, extracted.Err.Error())
+	}
+	return extracted.Text, nil
 }
 
 func (runner *ContainerRunner) CleanupDirs() {
@@ -624,15 +811,14 @@ func (runner *ContainerRunner) CommitLogs() error {
 	err = runner.ArvClient.Create("collections",
 		arvadosclient.Dict{
 			"collection": arvadosclient.Dict{
+				"is_trashed":    true,
 				"name":          "logs for " + runner.Container.UUID,
 				"manifest_text": mt}},
 		&response)
 	if err != nil {
 		return fmt.Errorf("While creating log collection: %v", err)
 	}
-
 	runner.LogsPDH = &response.PortableDataHash
-
 	return nil
 }
 
@@ -668,10 +854,10 @@ func (runner *ContainerRunner) ContainerToken() (string, error) {
 func (runner *ContainerRunner) UpdateContainerFinal() error {
 	update := arvadosclient.Dict{}
 	update["state"] = runner.finalState
+	if runner.LogsPDH != nil {
+		update["log"] = *runner.LogsPDH
+	}
 	if runner.finalState == "Complete" {
-		if runner.LogsPDH != nil {
-			update["log"] = *runner.LogsPDH
-		}
 		if runner.ExitCode != nil {
 			update["exit_code"] = *runner.ExitCode
 		}
@@ -731,6 +917,7 @@ func (runner *ContainerRunner) Run() (err error) {
 		checkErr(err)
 
 		if runner.finalState == "Queued" {
+			runner.CrunchLog.Close()
 			runner.UpdateContainerFinal()
 			return
 		}
@@ -763,6 +950,7 @@ func (runner *ContainerRunner) Run() (err error) {
 	// check for and/or load image
 	err = runner.LoadImage()
 	if err != nil {
+		runner.finalState = "Cancelled"
 		err = fmt.Errorf("While loading container image: %v", err)
 		return
 	}
@@ -770,6 +958,7 @@ func (runner *ContainerRunner) Run() (err error) {
 	// set up FUSE mount and binds
 	err = runner.SetupMounts()
 	if err != nil {
+		runner.finalState = "Cancelled"
 		err = fmt.Errorf("While setting up mounts: %v", err)
 		return
 	}
@@ -823,11 +1012,17 @@ func NewContainerRunner(api IArvadosClient,
 func main() {
 	statInterval := flag.Duration("crunchstat-interval", 10*time.Second, "sampling period for periodic resource usage reporting")
 	cgroupRoot := flag.String("cgroup-root", "/sys/fs/cgroup", "path to sysfs cgroup tree")
-	cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup")
+	cgroupParent := flag.String("cgroup-parent", "docker", "name of container's parent cgroup (ignored if -cgroup-parent-subsystem is used)")
+	cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
+	caCertsPath := flag.String("ca-certs", "", "Path to TLS root certificates")
 	flag.Parse()
 
 	containerId := flag.Arg(0)
 
+	if *caCertsPath != "" {
+		arvadosclient.CertFiles = []string{*caCertsPath}
+	}
+
 	api, err := arvadosclient.MakeArvadosClient()
 	if err != nil {
 		log.Fatalf("%s: %v", containerId, err)
@@ -835,7 +1030,7 @@ func main() {
 	api.Retries = 8
 
 	var kc *keepclient.KeepClient
-	kc, err = keepclient.MakeKeepClient(&api)
+	kc, err = keepclient.MakeKeepClient(api)
 	if err != nil {
 		log.Fatalf("%s: %v", containerId, err)
 	}
@@ -850,7 +1045,12 @@ func main() {
 	cr := NewContainerRunner(api, kc, docker, containerId)
 	cr.statInterval = *statInterval
 	cr.cgroupRoot = *cgroupRoot
-	cr.cgroupParent = *cgroupParent
+	cr.expectCgroupParent = *cgroupParent
+	if *cgroupParentSubsystem != "" {
+		p := findCgroup(*cgroupParentSubsystem)
+		cr.setCgroupParent = p
+		cr.expectCgroupParent = p
+	}
 
 	err = cr.Run()
 	if err != nil {