13497: Merge branch 'master' into 13497-controller

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

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 490d2cf480eee13c1c1d51fda7c3f2353a9ec2df..098c53f8a6a587816703ad6997ceb51eec7f0232 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)")
 
@@ -1007,6 +1014,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 +1090,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 +1122,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 {
@@ -1584,6 +1596,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 +1649,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