X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/ff30b2754f2517fff513f766398ec04eac14c11c..8d61948fecccfc60db2f18ba4daf7c01ddf3d3c8:/services/crunch-run/crunchrun.go

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 7da1beb20a..0b4eb2bcbf 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"git.curoverse.com/arvados.git/lib/crunchstat"
 	"git.curoverse.com/arvados.git/sdk/go/arvados"
 	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
 	"git.curoverse.com/arvados.git/sdk/go/keepclient"
@@ -17,6 +18,7 @@ import (
 	"os/exec"
 	"os/signal"
 	"path"
+	"path/filepath"
 	"strings"
 	"sync"
 	"syscall"
@@ -91,6 +93,24 @@ type ContainerRunner struct {
 	SigChan        chan os.Signal
 	ArvMountExit   chan error
 	finalState     string
+
+	statLogger   io.WriteCloser
+	statReporter *crunchstat.Reporter
+	statInterval time.Duration
+	cgroupRoot   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
@@ -102,7 +122,7 @@ func (runner *ContainerRunner) SetupSignals() {
 	signal.Notify(runner.SigChan, syscall.SIGQUIT)
 
 	go func(sig <-chan os.Signal) {
-		for _ = range sig {
+		for range sig {
 			if !runner.Cancelled {
 				runner.CancelLock.Lock()
 				runner.Cancelled = true
@@ -248,7 +268,8 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 			}
 		}
 
-		if mnt.Kind == "collection" {
+		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")
@@ -279,25 +300,47 @@ func (runner *ContainerRunner) SetupMounts() (err error) {
 				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))
 		}
 	}
 
@@ -366,6 +409,14 @@ func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
 				runner.CrunchLog.Printf("While closing stderr logs: %v", closeerr)
 			}
 
+			if runner.statReporter != nil {
+				runner.statReporter.Stop()
+				closeerr = runner.statLogger.Close()
+				if closeerr != nil {
+					runner.CrunchLog.Printf("While closing crunchstat logs: %v", closeerr)
+				}
+			}
+
 			runner.loggingDone <- true
 			close(runner.loggingDone)
 			return
@@ -373,6 +424,18 @@ func (runner *ContainerRunner) ProcessDockerAttach(containerReader io.Reader) {
 	}
 }
 
+func (runner *ContainerRunner) StartCrunchstat() {
+	runner.statLogger = NewThrottledLogger(runner.NewLogWriter("crunchstat"))
+	runner.statReporter = &crunchstat.Reporter{
+		CID:          runner.ContainerID,
+		Logger:       log.New(runner.statLogger, "", 0),
+		CgroupParent: runner.expectCgroupParent,
+		CgroupRoot:   runner.cgroupRoot,
+		PollPeriod:   runner.statInterval,
+	}
+	runner.statReporter.Start()
+}
+
 // AttachLogs connects the docker container stdout and stderr logs to the
 // Arvados logger which logs to Keep and the API server logs table.
 func (runner *ContainerRunner) AttachStreams() (err error) {
@@ -453,8 +516,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()
 }
@@ -752,6 +820,8 @@ func (runner *ContainerRunner) Run() (err error) {
 		return
 	}
 
+	runner.StartCrunchstat()
+
 	if runner.IsCancelled() {
 		return
 	}
@@ -792,6 +862,10 @@ 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 (ignored if -cgroup-parent-subsystem is used)")
+	cgroupParentSubsystem := flag.String("cgroup-parent-subsystem", "", "use current cgroup for given subsystem as parent cgroup for container")
 	flag.Parse()
 
 	containerId := flag.Arg(0)
@@ -816,6 +890,14 @@ func main() {
 	}
 
 	cr := NewContainerRunner(api, kc, docker, containerId)
+	cr.statInterval = *statInterval
+	cr.cgroupRoot = *cgroupRoot
+	cr.expectCgroupParent = *cgroupParent
+	if *cgroupParentSubsystem != "" {
+		p := findCgroup(*cgroupParentSubsystem)
+		cr.setCgroupParent = p
+		cr.expectCgroupParent = p
+	}
 
 	err = cr.Run()
 	if err != nil {