From: Lucas Di Pentima <ldipentima@veritasgenetics.com>
Date: Wed, 27 Jun 2018 20:43:58 +0000 (-0300)
Subject: 13219: Moves time out code from dispatch to crunch-run
X-Git-Tag: 1.2.0~94^2~2
X-Git-Url: https://git.arvados.org/arvados.git/commitdiff_plain/0765fb8b19dd7b76ec5d2d05edda3f4fd4347194

13219: Moves time out code from dispatch to crunch-run

Arvados-DCO-1.1-Signed-off-by: Lucas Di Pentima <ldipentima@veritasgenetics.com>
---

diff --git a/sdk/go/arvados/container.go b/sdk/go/arvados/container.go
index 1bdf089023..210ed9981c 100644
--- a/sdk/go/arvados/container.go
+++ b/sdk/go/arvados/container.go
@@ -20,7 +20,6 @@ type Container struct {
 	OutputPath           string               `json:"output_path"`
 	Priority             int                  `json:"priority"`
 	RuntimeConstraints   RuntimeConstraints   `json:"runtime_constraints"`
-	StartedAt            time.Time            `json:"started_at"`
 	State                ContainerState       `json:"state"`
 	SchedulingParameters SchedulingParameters `json:"scheduling_parameters"`
 }
diff --git a/sdk/go/dispatch/dispatch.go b/sdk/go/dispatch/dispatch.go
index ca2dbc48d4..3289c67b01 100644
--- a/sdk/go/dispatch/dispatch.go
+++ b/sdk/go/dispatch/dispatch.go
@@ -195,13 +195,6 @@ func (d *Dispatcher) checkListForUpdates(containers []arvados.Container, todo ma
 			case Queued:
 				tracker.close()
 			case Locked, Running:
-				if c.SchedulingParameters.MaxRunTime > 0 {
-					maxRunTime := time.Duration(c.SchedulingParameters.MaxRunTime) * time.Second
-					if time.Since(c.StartedAt) >= maxRunTime {
-						// Time's up, schedule container for cancellation
-						c.Priority = 0
-					}
-				}
 				tracker.update(c)
 			case Cancelled, Complete:
 				tracker.close()
diff --git a/services/crunch-run/crunchrun.go b/services/crunch-run/crunchrun.go
index 2f9ccf5246..adce853a53 100644
--- a/services/crunch-run/crunchrun.go
+++ b/services/crunch-run/crunchrun.go
@@ -1074,10 +1074,14 @@ func (runner *ContainerRunner) StartContainer() error {
 // WaitFinish waits for the container to terminate, capture the exit code, and
 // close the stdout/stderr logging.
 func (runner *ContainerRunner) WaitFinish() error {
+	var runTimeExceeded <-chan time.Time
 	runner.CrunchLog.Print("Waiting for container to finish")
 
 	waitOk, waitErr := runner.Docker.ContainerWait(context.TODO(), runner.ContainerID, dockercontainer.WaitConditionNotRunning)
 	arvMountExit := runner.ArvMountExit
+	if timeout := runner.Container.SchedulingParameters.MaxRunTime; timeout > 0 {
+		runTimeExceeded = time.After(time.Duration(timeout) * time.Second)
+	}
 	for {
 		select {
 		case waitBody := <-waitOk:
@@ -1098,6 +1102,11 @@ func (runner *ContainerRunner) WaitFinish() error {
 			// arvMountExit will always be ready now that
 			// it's closed, but that doesn't interest us.
 			arvMountExit = nil
+
+		case <-runTimeExceeded:
+			runner.CrunchLog.Printf("maximum run time exceeded. Stopping container.")
+			runner.stop(nil)
+			runTimeExceeded = nil
 		}
 	}
 }
diff --git a/services/crunch-run/crunchrun_test.go b/services/crunch-run/crunchrun_test.go
index c76682f1c6..8ad487d771 100644
--- a/services/crunch-run/crunchrun_test.go
+++ b/services/crunch-run/crunchrun_test.go
@@ -793,7 +793,7 @@ func (s *TestSuite) TestFullRunHello(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+	"runtime_constraints": {}
 }`, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -805,6 +805,26 @@ func (s *TestSuite) TestFullRunHello(c *C) {
 
 }
 
+func (s *TestSuite) TestRunTimeExceeded(c *C) {
+	api, _, _ := s.fullRunHelper(c, `{
+    "command": ["sleep", "3"],
+    "container_image": "d4ab34d3d4f8a72f5c4973051ae69fab+122",
+    "cwd": ".",
+    "environment": {},
+    "mounts": {"/tmp": {"kind": "tmp"} },
+    "output_path": "/tmp",
+    "priority": 1,
+	"runtime_constraints": {},
+	"scheduling_parameters":{"max_run_time": 1}
+}`, nil, 0, func(t *TestDockerClient) {
+		time.Sleep(3 * time.Second)
+		t.logWriter.Close()
+	})
+
+	c.Check(api.CalledWith("container.state", "Cancelled"), NotNil)
+	c.Check(api.Logs["crunch-run"].String(), Matches, "(?ms).*maximum run time exceeded.*")
+}
+
 func (s *TestSuite) TestCrunchstat(c *C) {
 	api, _, _ := s.fullRunHelper(c, `{
 		"command": ["sleep", "1"],