services/crunch-dispatch-local/crunch-dispatch-local_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "io"
  11         "net/http"
  12         "net/http/httptest"
  13         "os"
  14         "os/exec"
  15         "regexp"
  16         "testing"
  17         "time"
  18
  19         "git.curoverse.com/arvados.git/sdk/go/arvados"
  20         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  21         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
  22         "git.curoverse.com/arvados.git/sdk/go/dispatch"
  23         "github.com/sirupsen/logrus"
  24         . "gopkg.in/check.v1"
  25 )
  26
  27 // Gocheck boilerplate
  28 func Test(t *testing.T) {
  29         TestingT(t)
  30 }
  31
  32 var _ = Suite(&TestSuite{})
  33 var _ = Suite(&MockArvadosServerSuite{})
  34
  35 type TestSuite struct{}
  36 type MockArvadosServerSuite struct{}
  37
  38 var initialArgs []string
  39
  40 func (s *TestSuite) SetUpSuite(c *C) {
  41         initialArgs = os.Args
  42         arvadostest.StartAPI()
  43         runningCmds = make(map[string]*exec.Cmd)
  44         logrus.SetFormatter(&logrus.TextFormatter{DisableColors: true})
  45 }
  46
  47 func (s *TestSuite) TearDownSuite(c *C) {
  48         arvadostest.StopAPI()
  49 }
  50
  51 func (s *TestSuite) SetUpTest(c *C) {
  52         args := []string{"crunch-dispatch-local"}
  53         os.Args = args
  54 }
  55
  56 func (s *TestSuite) TearDownTest(c *C) {
  57         arvadostest.ResetEnv()
  58         os.Args = initialArgs
  59 }
  60
  61 func (s *MockArvadosServerSuite) TearDownTest(c *C) {
  62         arvadostest.ResetEnv()
  63 }
  64
  65 func (s *TestSuite) TestIntegration(c *C) {
  66         arv, err := arvadosclient.MakeArvadosClient()
  67         c.Assert(err, IsNil)
  68
  69         echo := "echo"
  70         crunchRunCommand = &echo
  71
  72         ctx, cancel := context.WithCancel(context.Background())
  73         dispatcher := dispatch.Dispatcher{
  74                 Arv:        arv,
  75                 PollPeriod: time.Second,
  76         }
  77
  78         startCmd := func(container arvados.Container, cmd *exec.Cmd) error {
  79                 dispatcher.UpdateState(container.UUID, "Running")
  80                 dispatcher.UpdateState(container.UUID, "Complete")
  81                 return cmd.Start()
  82         }
  83
  84         dispatcher.RunContainer = func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
  85                 (&LocalRun{startCmd, make(chan bool, 8), ctx}).run(d, c, s)
  86                 cancel()
  87         }
  88
  89         err = dispatcher.Run(ctx)
  90         c.Assert(err, Equals, context.Canceled)
  91
  92         // Wait for all running crunch jobs to complete / terminate
  93         waitGroup.Wait()
  94
  95         // There should be no queued containers now
  96         params := arvadosclient.Dict{
  97                 "filters": [][]string{{"state", "=", "Queued"}},
  98         }
  99         var containers arvados.ContainerList
 100         err = arv.List("containers", params, &containers)
 101         c.Check(err, IsNil)
 102         c.Assert(len(containers.Items), Equals, 0)
 103
 104         // Previously "Queued" container should now be in "Complete" state
 105         var container arvados.Container
 106         err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container)
 107         c.Check(err, IsNil)
 108         c.Check(string(container.State), Equals, "Complete")
 109 }
 110
 111 func (s *MockArvadosServerSuite) Test_APIErrorGettingContainers(c *C) {
 112         apiStubResponses := make(map[string]arvadostest.StubResponse)
 113         apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
 114
 115         testWithServerStub(c, apiStubResponses, "echo", "error getting count of containers")
 116 }
 117
 118 func (s *MockArvadosServerSuite) Test_APIErrorUpdatingContainerState(c *C) {
 119         apiStubResponses := make(map[string]arvadostest.StubResponse)
 120         apiStubResponses["/arvados/v1/containers"] =
 121                 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx1","State":"Queued","Priority":1}]}`)}
 122         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx1"] =
 123                 arvadostest.StubResponse{500, string(`{}`)}
 124
 125         testWithServerStub(c, apiStubResponses, "echo", "error locking container zzzzz-dz642-xxxxxxxxxxxxxx1")
 126 }
 127
 128 func (s *MockArvadosServerSuite) Test_ContainerStillInRunningAfterRun(c *C) {
 129         apiStubResponses := make(map[string]arvadostest.StubResponse)
 130         apiStubResponses["/arvados/v1/containers"] =
 131                 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2","State":"Queued","Priority":1}]}`)}
 132         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2/lock"] =
 133                 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Locked", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
 134         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2"] =
 135                 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
 136
 137         testWithServerStub(c, apiStubResponses, "echo",
 138                 `after \\"echo\\" process termination, container state for zzzzz-dz642-xxxxxxxxxxxxxx2 is \\"Running\\"; updating it to \\"Cancelled\\"`)
 139 }
 140
 141 func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {
 142         apiStubResponses := make(map[string]arvadostest.StubResponse)
 143         apiStubResponses["/arvados/v1/containers"] =
 144                 arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3","State":"Queued","Priority":1}]}`)}
 145
 146         apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx3/lock"] =
 147                 arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3", "state":"Locked", "priority":1}`)}
 148
 149         testWithServerStub(c, apiStubResponses, "nosuchcommand", `error starting \\"nosuchcommand\\" for zzzzz-dz642-xxxxxxxxxxxxxx3`)
 150 }
 151
 152 func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) {
 153         apiStubResponses["/arvados/v1/api_client_authorizations/current"] =
 154                 arvadostest.StubResponse{200, string(`{"uuid": "` + arvadostest.Dispatch1AuthUUID + `", "api_token": "xyz"}`)}
 155
 156         apiStub := arvadostest.ServerStub{apiStubResponses}
 157
 158         api := httptest.NewServer(&apiStub)
 159         defer api.Close()
 160
 161         arv := &arvadosclient.ArvadosClient{
 162                 Scheme:    "http",
 163                 ApiServer: api.URL[7:],
 164                 ApiToken:  "abc123",
 165                 Client:    &http.Client{Transport: &http.Transport{}},
 166                 Retries:   0,
 167         }
 168
 169         buf := bytes.NewBuffer(nil)
 170         logrus.SetOutput(io.MultiWriter(buf, os.Stderr))
 171         defer logrus.SetOutput(os.Stderr)
 172
 173         *crunchRunCommand = crunchCmd
 174
 175         ctx, cancel := context.WithCancel(context.Background())
 176         dispatcher := dispatch.Dispatcher{
 177                 Arv:        arv,
 178                 PollPeriod: time.Second / 20,
 179         }
 180
 181         startCmd := func(container arvados.Container, cmd *exec.Cmd) error {
 182                 dispatcher.UpdateState(container.UUID, "Running")
 183                 dispatcher.UpdateState(container.UUID, "Complete")
 184                 return cmd.Start()
 185         }
 186
 187         dispatcher.RunContainer = func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
 188                 (&LocalRun{startCmd, make(chan bool, 8), ctx}).run(d, c, s)
 189                 cancel()
 190         }
 191
 192         re := regexp.MustCompile(`(?ms).*` + expected + `.*`)
 193         go func() {
 194                 for i := 0; i < 80 && !re.MatchString(buf.String()); i++ {
 195                         time.Sleep(100 * time.Millisecond)
 196                 }
 197                 cancel()
 198         }()
 199
 200         err := dispatcher.Run(ctx)
 201         c.Assert(err, Equals, context.Canceled)
 202
 203         // Wait for all running crunch jobs to complete / terminate
 204         waitGroup.Wait()
 205
 206         c.Check(buf.String(), Matches, `(?ms).*`+expected+`.*`)
 207 }