12983: Periodically check for containerd

Arvados-DCO-1.1-Signed-off-by: Peter Amstutz <pamstutz@veritasgenetics.com>

diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index adce853a531e0f5012d88796613b30e4ff5a33d3..490d2cf480eee13c1c1d51fda7c3f2353a9ec2df 100644 (file)
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -32,6 +32,7 @@ import (
        "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
        "git.curoverse.com/arvados.git/sdk/go/keepclient"
        "git.curoverse.com/arvados.git/sdk/go/manifest"
+       "github.com/shirou/gopsutil/process"
        "golang.org/x/net/context"
 
        dockertypes "github.com/docker/docker/api/types"
@@ -141,9 +142,10 @@ type ContainerRunner struct {
        cStateLock sync.Mutex
        cCancelled bool // StopContainer() invoked
 
-       enableNetwork string // one of "default" or "always"
-       networkMode   string // passed through to HostConfig.NetworkMode
-       arvMountLog   *ThrottledLogger
+       enableNetwork   string // one of "default" or "always"
+       networkMode     string // passed through to HostConfig.NetworkMode
+       arvMountLog     *ThrottledLogger
+       checkContainerd bool
 }
 
 // setupSignals sets up signal handling to gracefully terminate the underlying
@@ -185,23 +187,27 @@ var errorBlacklist = []string{
 }
 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)")
 
+func (runner *ContainerRunner) runBrokenNodeHook() {
+       if *brokenNodeHook == "" {
+               runner.CrunchLog.Printf("No broken node hook provided, cannot mark node as broken.")
+       } else {
+               runner.CrunchLog.Printf("Running broken node hook %q", *brokenNodeHook)
+               // run killme script
+               c := exec.Command(*brokenNodeHook)
+               c.Stdout = runner.CrunchLog
+               c.Stderr = runner.CrunchLog
+               err := c.Run()
+               if err != nil {
+                       runner.CrunchLog.Printf("Error running broken node hook: %v", err)
+               }
+       }
+}
+
 func (runner *ContainerRunner) checkBrokenNode(goterr error) bool {
        for _, d := range errorBlacklist {
                if m, e := regexp.MatchString(d, goterr.Error()); m && e == nil {
                        runner.CrunchLog.Printf("Error suggests node is unable to run containers: %v", goterr)
-                       if *brokenNodeHook == "" {
-                               runner.CrunchLog.Printf("No broken node hook provided, cannot mark node as broken.")
-                       } else {
-                               runner.CrunchLog.Printf("Running broken node hook %q", *brokenNodeHook)
-                               // run killme script
-                               c := exec.Command(*brokenNodeHook)
-                               c.Stdout = runner.CrunchLog
-                               c.Stderr = runner.CrunchLog
-                               err := c.Run()
-                               if err != nil {
-                                       runner.CrunchLog.Printf("Error running broken node hook: %v", err)
-                               }
-                       }
+                       runner.runBrokenNodeHook()
                        return true
                }
        }
@@ -1071,6 +1077,27 @@ func (runner *ContainerRunner) StartContainer() error {
        return nil
 }
 
+// checkContainerd checks if "containerd" is present in the process list.
+func (runner *ContainerRunner) CheckContainerd() error {
+       if !runner.checkContainerd {
+               return nil
+       }
+       p, _ := process.Processes()
+       for _, i := range p {
+               e, _ := i.CmdlineSlice()
+               if len(e) > 0 {
+                       if strings.Index(e[0], "containerd") > -1 {
+                               return nil
+                       }
+               }
+       }
+
+       // Not found
+       runner.runBrokenNodeHook()
+       runner.stop(nil)
+       return fmt.Errorf("'containerd' not found in process list.")
+}
+
 // WaitFinish waits for the container to terminate, capture the exit code, and
 // close the stdout/stderr logging.
 func (runner *ContainerRunner) WaitFinish() error {
@@ -1082,6 +1109,27 @@ func (runner *ContainerRunner) WaitFinish() error {
        if timeout := runner.Container.SchedulingParameters.MaxRunTime; timeout > 0 {
                runTimeExceeded = time.After(time.Duration(timeout) * time.Second)
        }
+
+       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
+                                       return
+                               }
+                       case <-containerdGone:
+                               break
+                       }
+               }
+       }()
+
        for {
                select {
                case waitBody := <-waitOk:
@@ -1107,6 +1155,9 @@ func (runner *ContainerRunner) WaitFinish() error {
                        runner.CrunchLog.Printf("maximum run time exceeded. Stopping container.")
                        runner.stop(nil)
                        runTimeExceeded = nil
+
+               case err := <-containerdGone:
+                       return err
                }
        }
 }
@@ -1408,6 +1459,12 @@ func (runner *ContainerRunner) Run() (err error) {
                return
        }
 
+       // Sanity check that containerd is running.
+       err = runner.CheckContainerd()
+       if err != nil {
+               return
+       }
+
        // check for and/or load image
        err = runner.LoadImage()
        if err != nil {
@@ -1569,6 +1626,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.")
        flag.Parse()
 
        // Print version information if requested
@@ -1624,6 +1682,7 @@ func main() {
        cr.expectCgroupParent = *cgroupParent
        cr.enableNetwork = *enableNetwork
        cr.networkMode = *networkMode
+       cr.checkContainerd = *checkContainerd
        if *cgroupParentSubsystem != "" {
                p := findCgroup(*cgroupParentSubsystem)
                cr.setCgroupParent = p

diff --git a/services/crunch-run/logging_test.go b/services/crunch-run/logging_test.go
index 86f8cec04ae8037e37a1d4c9250216416f2f9bd6..13a171ae8416729cf67fd940a2170d871abc5bd1 100644 (file)
--- a/services/crunch-run/logging_test.go
+++ b/services/crunch-run/logging_test.go
@@ -83,7 +83,7 @@ func (s *LoggingTestSuite) TestWriteLogsLarge(c *C) {
        cr.CrunchLog.Print("Goodbye")
        cr.CrunchLog.Close()
 
-       c.Check(api.Calls > 1, Equals, true)
+       c.Check(api.Calls > 0, Equals, true)
        c.Check(api.Calls < 2000000, Equals, true)
 
        mt, err := cr.LogCollection.MarshalManifest(".")

diff --git a/vendor/vendor.json b/vendor/vendor.json
index a4f750b4c4d0445567ad20da7ac9408eb12a692d..f18d4e464cdf34ed86c0be1a4631aadc598179df 100644 (file)
--- a/vendor/vendor.json
+++ b/vendor/vendor.json
@@ -71,6 +71,12 @@
                        "revision": "d682213848ed68c0a260ca37d6dd5ace8423f5ba",
                        "revisionTime": "2017-12-05T20:32:29Z"
                },
+               {
+                       "checksumSHA1": "st4vb0GmDeoKbsfxdpNZ2MPl76M=",
+                       "path": "github.com/StackExchange/wmi",
+                       "revision": "cdffdb33acae0e14efff2628f9bae377b597840e",
+                       "revisionTime": "2018-04-12T20:51:11Z"
+               },
                {
                        "checksumSHA1": "spyv5/YFBjYyZLZa1U2LBfDR8PM=",
                        "path": "github.com/beorn7/perks/quantile",
@@ -233,6 +239,18 @@
                        "revision": "0ca9ea5df5451ffdf184b4428c902747c2c11cd7",
                        "revisionTime": "2017-03-27T23:54:44Z"
                },
+               {
+                       "checksumSHA1": "Kqv7bA4oJG0nPwQvGWDwGGaKONo=",
+                       "path": "github.com/go-ole/go-ole",
+                       "revision": "7a0fa49edf48165190530c675167e2f319a05268",
+                       "revisionTime": "2018-06-25T08:58:08Z"
+               },
+               {
+                       "checksumSHA1": "PArleDBtadu2qO4hJwHR8a3IOTA=",
+                       "path": "github.com/go-ole/go-ole/oleutil",
+                       "revision": "7a0fa49edf48165190530c675167e2f319a05268",
+                       "revisionTime": "2018-06-25T08:58:08Z"
+               },
                {
                        "checksumSHA1": "wn2shNJMwRZpvuvkf1s7h0wvqHI=",
                        "path": "github.com/gogo/protobuf/proto",
@@ -426,6 +444,48 @@
                        "revision": "1744e2970ca51c86172c8190fadad617561ed6e7",
                        "revisionTime": "2017-11-10T11:01:46Z"
                },
+               {
+                       "checksumSHA1": "q14d3C3xvWevU3dSv4P5K0+OSD0=",
+                       "path": "github.com/shirou/gopsutil/cpu",
+                       "revision": "63728fcf6b24475ecfea044e22242447666c2f52",
+                       "revisionTime": "2018-07-05T13:28:12Z"
+               },
+               {
+                       "checksumSHA1": "LZ9GloiGLTISmQ4dalK2XspH6Wo=",
+                       "path": "github.com/shirou/gopsutil/host",
+                       "revision": "63728fcf6b24475ecfea044e22242447666c2f52",
+                       "revisionTime": "2018-07-05T13:28:12Z"
+               },
+               {
+                       "checksumSHA1": "cyoqI0gryzjxGTkaAfyUqMiuUR0=",
+                       "path": "github.com/shirou/gopsutil/internal/common",
+                       "revision": "63728fcf6b24475ecfea044e22242447666c2f52",
+                       "revisionTime": "2018-07-05T13:28:12Z"
+               },
+               {
+                       "checksumSHA1": "vEQLjAO5T5K9zXblEMYdoaBZzj0=",
+                       "path": "github.com/shirou/gopsutil/mem",
+                       "revision": "63728fcf6b24475ecfea044e22242447666c2f52",
+                       "revisionTime": "2018-07-05T13:28:12Z"
+               },
+               {
+                       "checksumSHA1": "KMWFRa0DVpabo9d8euB4RYjUBQE=",
+                       "path": "github.com/shirou/gopsutil/net",
+                       "revision": "63728fcf6b24475ecfea044e22242447666c2f52",
+                       "revisionTime": "2018-07-05T13:28:12Z"
+               },
+               {
+                       "checksumSHA1": "fbO7c1gv1kSvWKOb/+5HUWFkBaA=",
+                       "path": "github.com/shirou/gopsutil/process",
+                       "revision": "63728fcf6b24475ecfea044e22242447666c2f52",
+                       "revisionTime": "2018-07-05T13:28:12Z"
+               },
+               {
+                       "checksumSHA1": "Nve7SpDmjsv6+rhkXAkfg/UQx94=",
+                       "path": "github.com/shirou/w32",
+                       "revision": "bb4de0191aa41b5507caa14b0650cdbddcd9280b",
+                       "revisionTime": "2016-09-30T03:27:40Z"
+               },
                {
                        "checksumSHA1": "8QeSG127zQqbA+YfkO1WkKx/iUI=",
                        "path": "github.com/src-d/gcfg",