X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/27fac89679511949fba6d5fb29eb905c579d2d97..6d95130da47af9fd0290d3c8f80a0364faf74957:/services/crunch-run/crunchrun_test.go

diff --git a/services/crunch-run/crunchrun_test.go b/services/crunch-run/crunchrun_test.go
index 8ee462586d..636926a8d0 100644
--- a/services/crunch-run/crunchrun_test.go
+++ b/services/crunch-run/crunchrun_test.go
@@ -24,10 +24,10 @@ import (
 	"testing"
 	"time"
 
-	"git.curoverse.com/arvados.git/sdk/go/arvados"
-	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
-	"git.curoverse.com/arvados.git/sdk/go/arvadostest"
-	"git.curoverse.com/arvados.git/sdk/go/manifest"
+	"git.arvados.org/arvados.git/sdk/go/arvados"
+	"git.arvados.org/arvados.git/sdk/go/arvadosclient"
+	"git.arvados.org/arvados.git/sdk/go/arvadostest"
+	"git.arvados.org/arvados.git/sdk/go/manifest"
 	"golang.org/x/net/context"
 
 	dockertypes "github.com/docker/docker/api/types"
@@ -47,6 +47,7 @@ var _ = Suite(&TestSuite{})
 type TestSuite struct {
 	client *arvados.Client
 	docker *TestDockerClient
+	runner *ContainerRunner
 }
 
 func (s *TestSuite) SetUpTest(c *C) {
@@ -103,6 +104,7 @@ type TestDockerClient struct {
 	api         *ArvTestClient
 	realTemp    string
 	calledWait  bool
+	ctrExited   bool
 }
 
 func NewTestDockerClient() *TestDockerClient {
@@ -176,6 +178,17 @@ func (t *TestDockerClient) ContainerWait(ctx context.Context, container string,
 	return body, err
 }
 
+func (t *TestDockerClient) ContainerInspect(ctx context.Context, id string) (c dockertypes.ContainerJSON, err error) {
+	c.ContainerJSONBase = &dockertypes.ContainerJSONBase{}
+	c.ID = "abcde"
+	if t.ctrExited {
+		c.State = &dockertypes.ContainerState{Status: "exited", Dead: true}
+	} else {
+		c.State = &dockertypes.ContainerState{Status: "running", Pid: 1234, Running: true}
+	}
+	return
+}
+
 func (t *TestDockerClient) ImageInspectWithRaw(ctx context.Context, image string) (dockertypes.ImageInspect, []byte, error) {
 	if t.exitCode == 2 {
 		return dockertypes.ImageInspect{}, nil, fmt.Errorf("Cannot connect to the Docker daemon at unix:///var/run/docker.sock. Is the docker daemon running?")
@@ -230,6 +243,7 @@ func (client *ArvTestClient) Create(resourceType string,
 		mt := parameters["collection"].(arvadosclient.Dict)["manifest_text"].(string)
 		outmap := output.(*arvados.Collection)
 		outmap.PortableDataHash = fmt.Sprintf("%x+%d", md5.Sum([]byte(mt)), len(mt))
+		outmap.UUID = fmt.Sprintf("zzzzz-4zz18-%15.15x", md5.Sum([]byte(mt)))
 	}
 
 	return nil
@@ -316,6 +330,10 @@ func (client *ArvTestClient) Update(resourceType string, uuid string, parameters
 		if parameters["container"].(arvadosclient.Dict)["state"] == "Running" {
 			client.WasSetRunning = true
 		}
+	} else if resourceType == "collections" {
+		mt := parameters["collection"].(arvadosclient.Dict)["manifest_text"].(string)
+		output.(*arvados.Collection).UUID = uuid
+		output.(*arvados.Collection).PortableDataHash = fmt.Sprintf("%x", md5.Sum([]byte(mt)))
 	}
 	return nil
 }
@@ -357,6 +375,10 @@ call:
 	return nil
 }
 
+func (client *KeepTestClient) LocalLocator(locator string) (string, error) {
+	return locator, nil
+}
+
 func (client *KeepTestClient) PutB(buf []byte) (string, int, error) {
 	client.Content = buf
 	return fmt.Sprintf("%x+%d", md5.Sum(buf), len(buf)), len(buf), nil
@@ -402,6 +424,10 @@ func (fw FileWrapper) Write([]byte) (int, error) {
 	return 0, errors.New("not implemented")
 }
 
+func (fw FileWrapper) Sync() error {
+	return errors.New("not implemented")
+}
+
 func (client *KeepTestClient) ManifestFileReader(m manifest.Manifest, filename string) (arvados.File, error) {
 	if filename == hwImageId+".tar" {
 		rdr := ioutil.NopCloser(&bytes.Buffer{})
@@ -416,10 +442,14 @@ func (client *KeepTestClient) ManifestFileReader(m manifest.Manifest, filename s
 }
 
 func (s *TestSuite) TestLoadImage(c *C) {
+	cr, err := NewContainerRunner(s.client, &ArvTestClient{},
+		&KeepTestClient{}, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
+	c.Assert(err, IsNil)
+
 	kc := &KeepTestClient{}
 	defer kc.Close()
-	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, kc, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
-	c.Assert(err, IsNil)
+	cr.ContainerArvClient = &ArvTestClient{}
+	cr.ContainerKeepClient = kc
 
 	_, err = cr.Docker.ImageRemove(nil, hwImageId, dockertypes.ImageRemoveOptions{})
 	c.Check(err, IsNil)
@@ -466,6 +496,9 @@ func (ArvErrorTestClient) Create(resourceType string,
 }
 
 func (ArvErrorTestClient) Call(method, resourceType, uuid, action string, parameters arvadosclient.Dict, output interface{}) error {
+	if method == "GET" && resourceType == "containers" && action == "auth" {
+		return nil
+	}
 	return errors.New("ArvError")
 }
 
@@ -498,6 +531,10 @@ func (*KeepErrorTestClient) PutB(buf []byte) (string, int, error) {
 	return "", 0, errors.New("KeepError")
 }
 
+func (*KeepErrorTestClient) LocalLocator(string) (string, error) {
+	return "", errors.New("KeepError")
+}
+
 type KeepReadErrorTestClient struct {
 	KeepTestClient
 }
@@ -526,8 +563,12 @@ func (s *TestSuite) TestLoadImageArvError(c *C) {
 	// (1) Arvados error
 	kc := &KeepTestClient{}
 	defer kc.Close()
-	cr, err := NewContainerRunner(s.client, ArvErrorTestClient{}, kc, nil, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
+	cr, err := NewContainerRunner(s.client, &ArvErrorTestClient{}, kc, nil, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
+
+	cr.ContainerArvClient = &ArvErrorTestClient{}
+	cr.ContainerKeepClient = &KeepTestClient{}
+
 	cr.Container.ContainerImage = hwPDH
 
 	err = cr.LoadImage()
@@ -536,8 +577,13 @@ func (s *TestSuite) TestLoadImageArvError(c *C) {
 
 func (s *TestSuite) TestLoadImageKeepError(c *C) {
 	// (2) Keep error
-	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, &KeepErrorTestClient{}, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
+	kc := &KeepErrorTestClient{}
+	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, kc, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
+
+	cr.ContainerArvClient = &ArvTestClient{}
+	cr.ContainerKeepClient = &KeepErrorTestClient{}
+
 	cr.Container.ContainerImage = hwPDH
 
 	err = cr.LoadImage()
@@ -547,19 +593,26 @@ func (s *TestSuite) TestLoadImageKeepError(c *C) {
 
 func (s *TestSuite) TestLoadImageCollectionError(c *C) {
 	// (3) Collection doesn't contain image
-	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, &KeepReadErrorTestClient{}, nil, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
+	kc := &KeepReadErrorTestClient{}
+	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, kc, nil, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
 	cr.Container.ContainerImage = otherPDH
 
+	cr.ContainerArvClient = &ArvTestClient{}
+	cr.ContainerKeepClient = &KeepReadErrorTestClient{}
+
 	err = cr.LoadImage()
 	c.Check(err.Error(), Equals, "First file in the container image collection does not end in .tar")
 }
 
 func (s *TestSuite) TestLoadImageKeepReadError(c *C) {
 	// (4) Collection doesn't contain image
-	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, &KeepReadErrorTestClient{}, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
+	kc := &KeepReadErrorTestClient{}
+	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, kc, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
 	cr.Container.ContainerImage = hwPDH
+	cr.ContainerArvClient = &ArvTestClient{}
+	cr.ContainerKeepClient = &KeepReadErrorTestClient{}
 
 	err = cr.LoadImage()
 	c.Check(err, NotNil)
@@ -607,6 +660,9 @@ func (s *TestSuite) TestRunContainer(c *C) {
 	cr, err := NewContainerRunner(s.client, &ArvTestClient{}, kc, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
 
+	cr.ContainerArvClient = &ArvTestClient{}
+	cr.ContainerKeepClient = &KeepTestClient{}
+
 	var logs TestLogs
 	cr.NewLogWriter = logs.NewTestLoggingWriter
 	cr.Container.ContainerImage = hwPDH
@@ -727,7 +783,9 @@ func (s *TestSuite) fullRunHelper(c *C, record string, extraMounts []string, exi
 	defer kc.Close()
 	cr, err = NewContainerRunner(s.client, api, kc, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
+	s.runner = cr
 	cr.statInterval = 100 * time.Millisecond
+	cr.containerWatchdogInterval = time.Second
 	am := &ArvMountCmdLine{}
 	cr.RunArvMount = am.ArvMountTest
 
@@ -748,8 +806,8 @@ func (s *TestSuite) fullRunHelper(c *C, record string, extraMounts []string, exi
 		}
 		return d, err
 	}
-	cr.MkArvClient = func(token string) (IArvadosClient, error) {
-		return &ArvTestClient{secretMounts: secretMounts}, nil
+	cr.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
+		return &ArvTestClient{secretMounts: secretMounts}, &KeepTestClient{}, nil, nil
 	}
 
 	if extraMounts != nil && len(extraMounts) > 0 {
@@ -767,7 +825,15 @@ func (s *TestSuite) fullRunHelper(c *C, record string, extraMounts []string, exi
 	}
 	if exitCode != 2 {
 		c.Check(api.WasSetRunning, Equals, true)
-		c.Check(api.Content[api.Calls-2]["container"].(arvadosclient.Dict)["log"], NotNil)
+		var lastupdate arvadosclient.Dict
+		for _, content := range api.Content {
+			if content["container"] != nil {
+				lastupdate = content["container"].(arvadosclient.Dict)
+			}
+		}
+		if lastupdate["log"] == nil {
+			c.Errorf("no container update with non-nil log -- updates were: %v", api.Content)
+		}
 	}
 
 	if err != nil {
@@ -789,7 +855,8 @@ func (s *TestSuite) TestFullRunHello(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -801,6 +868,68 @@ func (s *TestSuite) TestFullRunHello(c *C) {
 
 }
 
+func (s *TestSuite) TestRunAlreadyRunning(c *C) {
+	var ran bool
+	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},
+    "state": "Running"
+}`, nil, 2, func(t *TestDockerClient) {
+		ran = true
+	})
+
+	c.Check(api.CalledWith("container.state", "Cancelled"), IsNil)
+	c.Check(api.CalledWith("container.state", "Complete"), IsNil)
+	c.Check(ran, Equals, false)
+}
+
+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},
+    "state": "Locked"
+}`, 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) TestContainerWaitFails(c *C) {
+	api, _, _ := s.fullRunHelper(c, `{
+    "command": ["sleep", "3"],
+    "container_image": "d4ab34d3d4f8a72f5c4973051ae69fab+122",
+    "cwd": ".",
+    "mounts": {"/tmp": {"kind": "tmp"} },
+    "output_path": "/tmp",
+    "priority": 1,
+    "state": "Locked"
+}`, nil, 0, func(t *TestDockerClient) {
+		t.ctrExited = true
+		time.Sleep(10 * time.Second)
+		t.logWriter.Close()
+	})
+
+	c.Check(api.CalledWith("container.state", "Cancelled"), NotNil)
+	c.Check(api.Logs["crunch-run"].String(), Matches, "(?ms).*Container is not running.*")
+}
+
 func (s *TestSuite) TestCrunchstat(c *C) {
 	api, _, _ := s.fullRunHelper(c, `{
 		"command": ["sleep", "1"],
@@ -810,7 +939,8 @@ func (s *TestSuite) TestCrunchstat(c *C) {
 		"mounts": {"/tmp": {"kind": "tmp"} },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`, nil, 0, func(t *TestDockerClient) {
 		time.Sleep(time.Second)
 		t.logWriter.Close()
@@ -843,7 +973,8 @@ func (s *TestSuite) TestNodeInfoLog(c *C) {
 		"mounts": {"/tmp": {"kind": "tmp"} },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`, nil, 0,
 		func(t *TestDockerClient) {
 			time.Sleep(time.Second)
@@ -877,7 +1008,8 @@ func (s *TestSuite) TestContainerRecordLog(c *C) {
 		"mounts": {"/tmp": {"kind": "tmp"} },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`, nil, 0,
 		func(t *TestDockerClient) {
 			time.Sleep(time.Second)
@@ -900,7 +1032,8 @@ func (s *TestSuite) TestFullRunStderr(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 1, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello\n"))
 		t.logWriter.Write(dockerLog(2, "world\n"))
@@ -925,7 +1058,8 @@ func (s *TestSuite) TestFullRunDefaultCwd(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, t.cwd+"\n"))
 		t.logWriter.Close()
@@ -946,7 +1080,8 @@ func (s *TestSuite) TestFullRunSetCwd(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, t.cwd+"\n"))
 		t.logWriter.Close()
@@ -987,7 +1122,8 @@ func (s *TestSuite) testStopContainer(c *C, setup func(cr *ContainerRunner)) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`
 
 	rec := arvados.Container{}
@@ -1007,8 +1143,8 @@ func (s *TestSuite) testStopContainer(c *C, setup func(cr *ContainerRunner)) {
 	cr, err := NewContainerRunner(s.client, api, kc, s.docker, "zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 	c.Assert(err, IsNil)
 	cr.RunArvMount = func([]string, string) (*exec.Cmd, error) { return nil, nil }
-	cr.MkArvClient = func(token string) (IArvadosClient, error) {
-		return &ArvTestClient{}, nil
+	cr.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
+		return &ArvTestClient{}, &KeepTestClient{}, nil, nil
 	}
 	setup(cr)
 
@@ -1042,7 +1178,8 @@ func (s *TestSuite) TestFullRunSetEnv(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, t.env[0][7:]+"\n"))
 		t.logWriter.Close()
@@ -1080,6 +1217,8 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 	c.Assert(err, IsNil)
 	am := &ArvMountCmdLine{}
 	cr.RunArvMount = am.ArvMountTest
+	cr.ContainerArvClient = &ArvTestClient{}
+	cr.ContainerKeepClient = &KeepTestClient{}
 
 	realTemp, err := ioutil.TempDir("", "crunchrun_test1-")
 	c.Assert(err, IsNil)
@@ -1118,7 +1257,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 		cr.ArvMountPoint = ""
 		cr.Container.Mounts = make(map[string]arvados.Mount)
 		cr.Container.Mounts["/tmp"] = arvados.Mount{Kind: "tmp"}
-		cr.OutputPath = "/tmp"
+		cr.Container.OutputPath = "/tmp"
 		cr.statInterval = 5 * time.Second
 		err := cr.SetupMounts()
 		c.Check(err, IsNil)
@@ -1137,7 +1276,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 		cr.Container.Mounts = make(map[string]arvados.Mount)
 		cr.Container.Mounts["/out"] = arvados.Mount{Kind: "tmp"}
 		cr.Container.Mounts["/tmp"] = arvados.Mount{Kind: "tmp"}
-		cr.OutputPath = "/out"
+		cr.Container.OutputPath = "/out"
 
 		err := cr.SetupMounts()
 		c.Check(err, IsNil)
@@ -1155,7 +1294,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 		cr.ArvMountPoint = ""
 		cr.Container.Mounts = make(map[string]arvados.Mount)
 		cr.Container.Mounts["/tmp"] = arvados.Mount{Kind: "tmp"}
-		cr.OutputPath = "/tmp"
+		cr.Container.OutputPath = "/tmp"
 
 		apiflag := true
 		cr.Container.RuntimeConstraints.API = &apiflag
@@ -1179,7 +1318,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 		cr.Container.Mounts = map[string]arvados.Mount{
 			"/keeptmp": {Kind: "collection", Writable: true},
 		}
-		cr.OutputPath = "/keeptmp"
+		cr.Container.OutputPath = "/keeptmp"
 
 		os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
 
@@ -1201,7 +1340,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 			"/keepinp": {Kind: "collection", PortableDataHash: "59389a8f9ee9d399be35462a0f92541c+53"},
 			"/keepout": {Kind: "collection", Writable: true},
 		}
-		cr.OutputPath = "/keepout"
+		cr.Container.OutputPath = "/keepout"
 
 		os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", os.ModePerm)
 		os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
@@ -1227,7 +1366,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 			"/keepinp": {Kind: "collection", PortableDataHash: "59389a8f9ee9d399be35462a0f92541c+53"},
 			"/keepout": {Kind: "collection", Writable: true},
 		}
-		cr.OutputPath = "/keepout"
+		cr.Container.OutputPath = "/keepout"
 
 		os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", os.ModePerm)
 		os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
@@ -1308,7 +1447,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 			"/tmp":     {Kind: "tmp"},
 			"/tmp/foo": {Kind: "collection"},
 		}
-		cr.OutputPath = "/tmp"
+		cr.Container.OutputPath = "/tmp"
 
 		os.MkdirAll(realTemp+"/keep1/tmp0", os.ModePerm)
 
@@ -1338,7 +1477,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 				Path:             "baz",
 				Writable:         true},
 		}
-		cr.OutputPath = "/tmp"
+		cr.Container.OutputPath = "/tmp"
 
 		os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541c+53", os.ModePerm)
 		os.MkdirAll(realTemp+"/keep1/by_id/59389a8f9ee9d399be35462a0f92541d+53/baz", os.ModePerm)
@@ -1367,7 +1506,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 			"/tmp":     {Kind: "tmp"},
 			"/tmp/foo": {Kind: "tmp"},
 		}
-		cr.OutputPath = "/tmp"
+		cr.Container.OutputPath = "/tmp"
 
 		err := cr.SetupMounts()
 		c.Check(err, NotNil)
@@ -1415,7 +1554,7 @@ func (s *TestSuite) TestSetupMounts(c *C) {
 				Path:   "/",
 			},
 		}
-		cr.OutputPath = "/tmp"
+		cr.Container.OutputPath = "/tmp"
 
 		err := cr.SetupMounts()
 		c.Check(err, IsNil)
@@ -1462,17 +1601,18 @@ func (s *TestSuite) TestStdout(c *C) {
 		"mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "file", "path": "/tmp/a/b/c.out"} },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
-	api, _, _ := s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
+	api, cr, _ := s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, t.env[0][7:]+"\n"))
 		t.logWriter.Close()
 	})
 
 	c.Check(api.CalledWith("container.exit_code", 0), NotNil)
 	c.Check(api.CalledWith("container.state", "Complete"), NotNil)
-	c.Check(api.CalledWith("collection.manifest_text", "./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out\n"), NotNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", "./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out\n"), NotNil)
 }
 
 // Used by the TestStdoutWithWrongPath*()
@@ -1491,8 +1631,8 @@ func (s *TestSuite) stdoutErrorRunHelper(c *C, record string, fn func(t *TestDoc
 	c.Assert(err, IsNil)
 	am := &ArvMountCmdLine{}
 	cr.RunArvMount = am.ArvMountTest
-	cr.MkArvClient = func(token string) (IArvadosClient, error) {
-		return &ArvTestClient{}, nil
+	cr.MkArvClient = func(token string) (IArvadosClient, IKeepClient, *arvados.Client, error) {
+		return &ArvTestClient{}, &KeepTestClient{}, nil, nil
 	}
 
 	err = cr.Run()
@@ -1502,7 +1642,8 @@ func (s *TestSuite) stdoutErrorRunHelper(c *C, record string, fn func(t *TestDoc
 func (s *TestSuite) TestStdoutWithWrongPath(c *C) {
 	_, _, err := s.stdoutErrorRunHelper(c, `{
     "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "file", "path":"/tmpa.out"} },
-    "output_path": "/tmp"
+    "output_path": "/tmp",
+    "state": "Locked"
 }`, func(t *TestDockerClient) {})
 
 	c.Check(err, NotNil)
@@ -1512,7 +1653,8 @@ func (s *TestSuite) TestStdoutWithWrongPath(c *C) {
 func (s *TestSuite) TestStdoutWithWrongKindTmp(c *C) {
 	_, _, err := s.stdoutErrorRunHelper(c, `{
     "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "tmp", "path":"/tmp/a.out"} },
-    "output_path": "/tmp"
+    "output_path": "/tmp",
+    "state": "Locked"
 }`, func(t *TestDockerClient) {})
 
 	c.Check(err, NotNil)
@@ -1522,7 +1664,8 @@ func (s *TestSuite) TestStdoutWithWrongKindTmp(c *C) {
 func (s *TestSuite) TestStdoutWithWrongKindCollection(c *C) {
 	_, _, err := s.stdoutErrorRunHelper(c, `{
     "mounts": {"/tmp": {"kind": "tmp"}, "stdout": {"kind": "collection", "path":"/tmp/a.out"} },
-    "output_path": "/tmp"
+    "output_path": "/tmp",
+    "state": "Locked"
 }`, func(t *TestDockerClient) {})
 
 	c.Check(err, NotNil)
@@ -1540,7 +1683,8 @@ func (s *TestSuite) TestFullRunWithAPI(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {"API": true}
+    "runtime_constraints": {"API": true},
+    "state": "Locked"
 }`, nil, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, t.env[1][17:]+"\n"))
 		t.logWriter.Close()
@@ -1563,7 +1707,8 @@ func (s *TestSuite) TestFullRunSetOutput(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {"API": true}
+    "runtime_constraints": {"API": true},
+    "state": "Locked"
 }`, nil, 0, func(t *TestDockerClient) {
 		t.api.Container.Output = "d4ab34d3d4f8a72f5c4973051ae69fab+122"
 		t.logWriter.Close()
@@ -1590,19 +1735,20 @@ func (s *TestSuite) TestStdoutWithExcludeFromOutputMountPointUnderOutputDir(c *C
     },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
 	extraMounts := []string{"a3e8f74c6f101eae01fa08bfb4e49b3a+54"}
 
-	api, _, _ := s.fullRunHelper(c, helperRecord, extraMounts, 0, func(t *TestDockerClient) {
+	api, cr, _ := s.fullRunHelper(c, helperRecord, extraMounts, 0, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, t.env[0][7:]+"\n"))
 		t.logWriter.Close()
 	})
 
 	c.Check(api.CalledWith("container.exit_code", 0), NotNil)
 	c.Check(api.CalledWith("container.state", "Complete"), NotNil)
-	c.Check(api.CalledWith("collection.manifest_text", "./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out\n"), NotNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", "./a/b 307372fa8fd5c146b22ae7a45b49bc31+6 0:6:c.out\n"), NotNil)
 }
 
 func (s *TestSuite) TestStdoutWithMultipleMountPointsUnderOutputDir(c *C) {
@@ -1621,7 +1767,8 @@ func (s *TestSuite) TestStdoutWithMultipleMountPointsUnderOutputDir(c *C) {
     },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
 	extraMounts := []string{
@@ -1675,7 +1822,8 @@ func (s *TestSuite) TestStdoutWithMountPointsUnderOutputDirDenormalizedManifest(
     },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
 	extraMounts := []string{
@@ -1710,11 +1858,12 @@ func (s *TestSuite) TestOutputError(c *C) {
 		"cwd": "/bin",
 		"environment": {"FROBIZ": "bilbo"},
 		"mounts": {
-        "/tmp": {"kind": "tmp"}
-    },
+			"/tmp": {"kind": "tmp"}
+		},
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
 	extraMounts := []string{}
@@ -1740,7 +1889,8 @@ func (s *TestSuite) TestStdinCollectionMountPoint(c *C) {
     },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
 	extraMounts := []string{
@@ -1779,7 +1929,8 @@ func (s *TestSuite) TestStdinJsonMountPoint(c *C) {
     },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
 	api, _, _ := s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
@@ -1802,7 +1953,7 @@ func (s *TestSuite) TestStdinJsonMountPoint(c *C) {
 }
 
 func (s *TestSuite) TestStderrMount(c *C) {
-	api, _, _ := s.fullRunHelper(c, `{
+	api, cr, _ := s.fullRunHelper(c, `{
     "command": ["/bin/sh", "-c", "echo hello;exit 1"],
     "container_image": "d4ab34d3d4f8a72f5c4973051ae69fab+122",
     "cwd": ".",
@@ -1812,7 +1963,8 @@ func (s *TestSuite) TestStderrMount(c *C) {
                "stderr": {"kind": "file", "path": "/tmp/b/err.txt"}},
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 1, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello\n"))
 		t.logWriter.Write(dockerLog(2, "oops\n"))
@@ -1824,7 +1976,7 @@ func (s *TestSuite) TestStderrMount(c *C) {
 	c.Check(final["container"].(arvadosclient.Dict)["exit_code"], Equals, 1)
 	c.Check(final["container"].(arvadosclient.Dict)["log"], NotNil)
 
-	c.Check(api.CalledWith("collection.manifest_text", "./a b1946ac92492d2347c6235b4d2611184+6 0:6:out.txt\n./b 38af5c54926b620264ab1501150cf189+5 0:5:err.txt\n"), NotNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", "./a b1946ac92492d2347c6235b4d2611184+6 0:6:out.txt\n./b 38af5c54926b620264ab1501150cf189+5 0:5:err.txt\n"), NotNil)
 }
 
 func (s *TestSuite) TestNumberRoundTrip(c *C) {
@@ -1862,7 +2014,8 @@ exec echo killme
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 2, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -1887,7 +2040,8 @@ func (s *TestSuite) TestFullBrokenDocker2(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 2, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -1895,7 +2049,7 @@ func (s *TestSuite) TestFullBrokenDocker2(c *C) {
 
 	c.Check(api.CalledWith("container.state", "Queued"), NotNil)
 	c.Check(api.Logs["crunch-run"].String(), Matches, "(?ms).*unable to run containers.*")
-	c.Check(api.Logs["crunch-run"].String(), Matches, "(?ms).*No broken node hook.*")
+	c.Check(api.Logs["crunch-run"].String(), Matches, "(?ms).*Writing /var/lock/crunch-run-broken to mark node as broken.*")
 }
 
 func (s *TestSuite) TestFullBrokenDocker3(c *C) {
@@ -1910,7 +2064,8 @@ func (s *TestSuite) TestFullBrokenDocker3(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 3, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -1932,7 +2087,8 @@ func (s *TestSuite) TestBadCommand1(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 4, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -1954,7 +2110,8 @@ func (s *TestSuite) TestBadCommand2(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 5, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -1976,7 +2133,8 @@ func (s *TestSuite) TestBadCommand3(c *C) {
     "mounts": {"/tmp": {"kind": "tmp"} },
     "output_path": "/tmp",
     "priority": 1,
-    "runtime_constraints": {}
+    "runtime_constraints": {},
+    "state": "Locked"
 }`, nil, 6, func(t *TestDockerClient) {
 		t.logWriter.Write(dockerLog(1, "hello world\n"))
 		t.logWriter.Close()
@@ -2000,10 +2158,11 @@ func (s *TestSuite) TestSecretTextMountPoint(c *C) {
                 },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
-	api, _, _ := s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
+	api, cr, _ := s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
 		content, err := ioutil.ReadFile(t.realTemp + "/tmp2/secret.conf")
 		c.Check(err, IsNil)
 		c.Check(content, DeepEquals, []byte("mypassword"))
@@ -2012,8 +2171,8 @@ func (s *TestSuite) TestSecretTextMountPoint(c *C) {
 
 	c.Check(api.CalledWith("container.exit_code", 0), NotNil)
 	c.Check(api.CalledWith("container.state", "Complete"), NotNil)
-	c.Check(api.CalledWith("collection.manifest_text", ". 34819d7beeabb9260a5c854bc85b3e44+10 0:10:secret.conf\n"), NotNil)
-	c.Check(api.CalledWith("collection.manifest_text", ""), IsNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ". 34819d7beeabb9260a5c854bc85b3e44+10 0:10:secret.conf\n"), NotNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ""), IsNil)
 
 	// under secret mounts, not captured in output
 	helperRecord = `{
@@ -2028,10 +2187,11 @@ func (s *TestSuite) TestSecretTextMountPoint(c *C) {
                 },
 		"output_path": "/tmp",
 		"priority": 1,
-		"runtime_constraints": {}
+		"runtime_constraints": {},
+		"state": "Locked"
 	}`
 
-	api, _, _ = s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
+	api, cr, _ = s.fullRunHelper(c, helperRecord, nil, 0, func(t *TestDockerClient) {
 		content, err := ioutil.ReadFile(t.realTemp + "/tmp2/secret.conf")
 		c.Check(err, IsNil)
 		c.Check(content, DeepEquals, []byte("mypassword"))
@@ -2040,6 +2200,14 @@ func (s *TestSuite) TestSecretTextMountPoint(c *C) {
 
 	c.Check(api.CalledWith("container.exit_code", 0), NotNil)
 	c.Check(api.CalledWith("container.state", "Complete"), NotNil)
-	c.Check(api.CalledWith("collection.manifest_text", ". 34819d7beeabb9260a5c854bc85b3e44+10 0:10:secret.conf\n"), IsNil)
-	c.Check(api.CalledWith("collection.manifest_text", ""), NotNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ". 34819d7beeabb9260a5c854bc85b3e44+10 0:10:secret.conf\n"), IsNil)
+	c.Check(cr.ContainerArvClient.(*ArvTestClient).CalledWith("collection.manifest_text", ""), NotNil)
+}
+
+type FakeProcess struct {
+	cmdLine []string
+}
+
+func (fp FakeProcess) CmdlineSlice() ([]string, error) {
+	return fp.cmdLine, nil
 }