10181: Add Size method to arvados.CollectionFileSystem.

[arvados.git] / services / crunch-run / crunchrun.go

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 490d2cf480eee13c1c1d51fda7c3f2353a9ec2df..730194b8251f9397828fa660d1e1c15d6ae5f63e 100644 (file)
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -84,6 +84,10 @@ type ThinDockerClient interface {
        ImageRemove(ctx context.Context, image string, options dockertypes.ImageRemoveOptions) ([]dockertypes.ImageDeleteResponseItem, error)
 }
 
+type PsProcess interface {
+       CmdlineSlice() ([]string, error)
+}
+
 // ContainerRunner is the main stateful struct used for a single execution of a
 // container.
 type ContainerRunner struct {
@@ -119,6 +123,8 @@ type ContainerRunner struct {
        finalState    string
        parentTemp    string
 
+       ListProcesses func() ([]PsProcess, error)
+
        statLogger       io.WriteCloser
        statReporter     *crunchstat.Reporter
        hoststatLogger   io.WriteCloser
@@ -145,7 +151,7 @@ type ContainerRunner struct {
        enableNetwork   string // one of "default" or "always"
        networkMode     string // passed through to HostConfig.NetworkMode
        arvMountLog     *ThrottledLogger
-       checkContainerd bool
+       checkContainerd time.Duration
 }
 
 // setupSignals sets up signal handling to gracefully terminate the underlying
@@ -184,6 +190,7 @@ func (runner *ContainerRunner) stop(sig os.Signal) {
 var errorBlacklist = []string{
        "(?ms).*[Cc]annot connect to the Docker daemon.*",
        "(?ms).*oci runtime error.*starting container process.*container init.*mounting.*to rootfs.*no such file or directory.*",
+       "(?ms).*grpc: the connection is unavailable.*",
 }
 var brokenNodeHook *string = flag.String("broken-node-hook", "", "Script to run if node is detected to be broken (for example, Docker daemon is not running)")
 
@@ -735,6 +742,7 @@ func (runner *ContainerRunner) startCrunchstat() error {
                CgroupParent: runner.expectCgroupParent,
                CgroupRoot:   runner.cgroupRoot,
                PollPeriod:   runner.statInterval,
+               TempDir:      runner.parentTemp,
        }
        runner.statReporter.Start()
        return nil
@@ -1007,6 +1015,10 @@ func (runner *ContainerRunner) CreateContainer() error {
        runner.ContainerConfig.Volumes = runner.Volumes
 
        maxRAM := int64(runner.Container.RuntimeConstraints.RAM)
+       if maxRAM < 4*1024*1024 {
+               // Docker daemon won't let you set a limit less than 4 MiB
+               maxRAM = 4 * 1024 * 1024
+       }
        runner.HostConfig = dockercontainer.HostConfig{
                Binds: runner.Binds,
                LogConfig: dockercontainer.LogConfig{
@@ -1079,10 +1091,10 @@ func (runner *ContainerRunner) StartContainer() error {
 
 // checkContainerd checks if "containerd" is present in the process list.
 func (runner *ContainerRunner) CheckContainerd() error {
-       if !runner.checkContainerd {
+       if runner.checkContainerd == 0 {
                return nil
        }
-       p, _ := process.Processes()
+       p, _ := runner.ListProcesses()
        for _, i := range p {
                e, _ := i.CmdlineSlice()
                if len(e) > 0 {
@@ -1111,24 +1123,25 @@ func (runner *ContainerRunner) WaitFinish() error {
        }
 
        containerdGone := make(chan error)
-       defer func() {
-               close(containerdGone)
-       }()
-       go func() {
-               ticker := time.NewTicker(time.Duration(60 * time.Second))
-               defer ticker.Stop()
-               for {
-                       select {
-                       case <-ticker.C:
-                               if ck := runner.CheckContainerd(); ck != nil {
-                                       containerdGone <- ck
+       defer close(containerdGone)
+       if runner.checkContainerd > 0 {
+               go func() {
+                       ticker := time.NewTicker(time.Duration(runner.checkContainerd))
+                       defer ticker.Stop()
+                       for {
+                               select {
+                               case <-ticker.C:
+                                       if ck := runner.CheckContainerd(); ck != nil {
+                                               containerdGone <- ck
+                                               return
+                                       }
+                               case <-containerdGone:
+                                       // Channel closed, quit goroutine
                                        return
                                }
-                       case <-containerdGone:
-                               break
                        }
-               }
-       }()
+               }()
+       }
 
        for {
                select {
@@ -1415,11 +1428,11 @@ func (runner *ContainerRunner) Run() (err error) {
                // hasn't already been assigned when Run() returns,
                // this cleanup func will cause Run() to return the
                // first non-nil error that is passed to checkErr().
-               checkErr := func(e error) {
+               checkErr := func(errorIn string, e error) {
                        if e == nil {
                                return
                        }
-                       runner.CrunchLog.Print(e)
+                       runner.CrunchLog.Printf("error in %s: %v", errorIn, e)
                        if err == nil {
                                err = e
                        }
@@ -1430,7 +1443,7 @@ func (runner *ContainerRunner) Run() (err error) {
                }
 
                // Log the error encountered in Run(), if any
-               checkErr(err)
+               checkErr("Run", err)
 
                if runner.finalState == "Queued" {
                        runner.UpdateContainerFinal()
@@ -1443,10 +1456,10 @@ func (runner *ContainerRunner) Run() (err error) {
                        // capture partial output and write logs
                }
 
-               checkErr(runner.CaptureOutput())
-               checkErr(runner.stopHoststat())
-               checkErr(runner.CommitLogs())
-               checkErr(runner.UpdateContainerFinal())
+               checkErr("CaptureOutput", runner.CaptureOutput())
+               checkErr("stopHoststat", runner.stopHoststat())
+               checkErr("CommitLogs", runner.CommitLogs())
+               checkErr("UpdateContainerFinal", runner.UpdateContainerFinal())
        }()
 
        err = runner.fetchContainerRecord()
@@ -1584,6 +1597,17 @@ func NewContainerRunner(client *arvados.Client, api IArvadosClient, kc IKeepClie
        cr.NewLogWriter = cr.NewArvLogWriter
        cr.RunArvMount = cr.ArvMountCmd
        cr.MkTempDir = ioutil.TempDir
+       cr.ListProcesses = func() ([]PsProcess, error) {
+               pr, err := process.Processes()
+               if err != nil {
+                       return nil, err
+               }
+               ps := make([]PsProcess, len(pr))
+               for i, j := range pr {
+                       ps[i] = j
+               }
+               return ps, nil
+       }
        cr.MkArvClient = func(token string) (IArvadosClient, error) {
                cl, err := arvadosclient.MakeArvadosClient()
                if err != nil {
@@ -1626,7 +1650,7 @@ func main() {
        `)
        memprofile := flag.String("memprofile", "", "write memory profile to `file` after running container")
        getVersion := flag.Bool("version", false, "Print version information and exit.")
-       checkContainerd := flag.Bool("check-containerd", false, "Periodically check if (docker-)containerd is running, cancel if missing.")
+       checkContainerd := flag.Duration("check-containerd", 60*time.Second, "Periodic check if (docker-)containerd is running (use 0s to disable).")
        flag.Parse()
 
        // Print version information if requested