1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
19 "git.arvados.org/arvados.git/sdk/go/arvados"
20 "git.arvados.org/arvados.git/sdk/go/arvadosclient"
21 "git.arvados.org/arvados.git/sdk/go/arvadostest"
22 "git.arvados.org/arvados.git/sdk/go/dispatch"
23 "github.com/sirupsen/logrus"
27 // Gocheck boilerplate
28 func Test(t *testing.T) {
32 var _ = Suite(&TestSuite{})
33 var _ = Suite(&MockArvadosServerSuite{})
35 type TestSuite struct{}
36 type MockArvadosServerSuite struct{}
38 var initialArgs []string
40 func (s *TestSuite) SetUpSuite(c *C) {
42 arvadostest.StartAPI()
43 runningCmds = make(map[string]*exec.Cmd)
44 logrus.SetFormatter(&logrus.TextFormatter{DisableColors: true})
47 func (s *TestSuite) TearDownSuite(c *C) {
51 func (s *TestSuite) SetUpTest(c *C) {
52 args := []string{"crunch-dispatch-local"}
56 func (s *TestSuite) TearDownTest(c *C) {
57 arvadostest.ResetEnv()
61 func (s *MockArvadosServerSuite) TearDownTest(c *C) {
62 arvadostest.ResetEnv()
65 func (s *TestSuite) TestIntegration(c *C) {
66 arv, err := arvadosclient.MakeArvadosClient()
70 crunchRunCommand = &echo
72 ctx, cancel := context.WithCancel(context.Background())
73 dispatcher := dispatch.Dispatcher{
75 PollPeriod: time.Second,
78 startCmd := func(container arvados.Container, cmd *exec.Cmd) error {
79 dispatcher.UpdateState(container.UUID, "Running")
80 dispatcher.UpdateState(container.UUID, "Complete")
84 cl := arvados.Cluster{Containers: arvados.ContainersConfig{RuntimeEngine: "docker"}}
86 dispatcher.RunContainer = func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
87 (&LocalRun{startCmd, make(chan bool, 8), ctx, &cl}).run(d, c, s)
91 err = dispatcher.Run(ctx)
92 c.Assert(err, Equals, context.Canceled)
94 // Wait for all running crunch jobs to complete / terminate
97 // There should be no queued containers now
98 params := arvadosclient.Dict{
99 "filters": [][]string{{"state", "=", "Queued"}},
101 var containers arvados.ContainerList
102 err = arv.List("containers", params, &containers)
104 c.Assert(len(containers.Items), Equals, 0)
106 // Previously "Queued" container should now be in "Complete" state
107 var container arvados.Container
108 err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container)
110 c.Check(string(container.State), Equals, "Complete")
113 func (s *MockArvadosServerSuite) Test_APIErrorGettingContainers(c *C) {
114 apiStubResponses := make(map[string]arvadostest.StubResponse)
115 apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
117 testWithServerStub(c, apiStubResponses, "echo", "error getting count of containers")
120 func (s *MockArvadosServerSuite) Test_APIErrorUpdatingContainerState(c *C) {
121 apiStubResponses := make(map[string]arvadostest.StubResponse)
122 apiStubResponses["/arvados/v1/containers"] =
123 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx1","State":"Queued","Priority":1}]}`)}
124 apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx1"] =
125 arvadostest.StubResponse{500, string(`{}`)}
127 testWithServerStub(c, apiStubResponses, "echo", "error locking container zzzzz-dz642-xxxxxxxxxxxxxx1")
130 func (s *MockArvadosServerSuite) Test_ContainerStillInRunningAfterRun(c *C) {
131 apiStubResponses := make(map[string]arvadostest.StubResponse)
132 apiStubResponses["/arvados/v1/containers"] =
133 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2","State":"Queued","Priority":1}]}`)}
134 apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2/lock"] =
135 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Locked", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
136 apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2"] =
137 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
139 testWithServerStub(c, apiStubResponses, "echo",
140 `after \\"echo\\" process termination, container state for zzzzz-dz642-xxxxxxxxxxxxxx2 is \\"Running\\"; updating it to \\"Cancelled\\"`)
143 func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {
144 apiStubResponses := make(map[string]arvadostest.StubResponse)
145 apiStubResponses["/arvados/v1/containers"] =
146 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3","State":"Queued","Priority":1}]}`)}
148 apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx3/lock"] =
149 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3", "state":"Locked", "priority":1}`)}
151 testWithServerStub(c, apiStubResponses, "nosuchcommand", `error starting \\"nosuchcommand\\" for zzzzz-dz642-xxxxxxxxxxxxxx3`)
154 func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) {
155 apiStubResponses["/arvados/v1/api_client_authorizations/current"] =
156 arvadostest.StubResponse{200, string(`{"uuid": "` + arvadostest.Dispatch1AuthUUID + `", "api_token": "xyz"}`)}
158 apiStub := arvadostest.ServerStub{apiStubResponses}
160 api := httptest.NewServer(&apiStub)
163 arv := &arvadosclient.ArvadosClient{
165 ApiServer: api.URL[7:],
167 Client: &http.Client{Transport: &http.Transport{}},
171 buf := bytes.NewBuffer(nil)
172 logrus.SetOutput(io.MultiWriter(buf, os.Stderr))
173 defer logrus.SetOutput(os.Stderr)
175 *crunchRunCommand = crunchCmd
177 ctx, cancel := context.WithCancel(context.Background())
178 dispatcher := dispatch.Dispatcher{
180 PollPeriod: time.Second / 20,
183 startCmd := func(container arvados.Container, cmd *exec.Cmd) error {
184 dispatcher.UpdateState(container.UUID, "Running")
185 dispatcher.UpdateState(container.UUID, "Complete")
189 cl := arvados.Cluster{Containers: arvados.ContainersConfig{RuntimeEngine: "docker"}}
191 dispatcher.RunContainer = func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
192 (&LocalRun{startCmd, make(chan bool, 8), ctx, &cl}).run(d, c, s)
196 re := regexp.MustCompile(`(?ms).*` + expected + `.*`)
198 for i := 0; i < 80 && !re.MatchString(buf.String()); i++ {
199 time.Sleep(100 * time.Millisecond)
204 err := dispatcher.Run(ctx)
205 c.Assert(err, Equals, context.Canceled)
207 // Wait for all running crunch jobs to complete / terminate
210 c.Check(buf.String(), Matches, `(?ms).*`+expected+`.*`)