X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e5de95c8d7040a2810238e069dd18403765ce056..396cb080346a46a94508b2938ef6d916f4d1778f:/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go diff --git a/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go b/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go index a823755379..499f6d5d72 100644 --- a/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go +++ b/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go @@ -7,6 +7,7 @@ package main import ( "bytes" "context" + "errors" "fmt" "io" "io/ioutil" @@ -19,6 +20,7 @@ import ( "testing" "time" + "git.curoverse.com/arvados.git/lib/dispatchcloud" "git.curoverse.com/arvados.git/sdk/go/arvados" "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/arvadostest" @@ -31,84 +33,77 @@ func Test(t *testing.T) { TestingT(t) } -var _ = Suite(&TestSuite{}) -var _ = Suite(&MockArvadosServerSuite{}) +var _ = Suite(&IntegrationSuite{}) +var _ = Suite(&StubbedSuite{}) -type TestSuite struct{} -type MockArvadosServerSuite struct{} - -var initialArgs []string - -func (s *TestSuite) SetUpSuite(c *C) { - initialArgs = os.Args +type IntegrationSuite struct { + disp Dispatcher + slurm slurmFake } -func (s *TestSuite) TearDownSuite(c *C) { -} - -func (s *TestSuite) SetUpTest(c *C) { - args := []string{"crunch-dispatch-slurm"} - os.Args = args - +func (s *IntegrationSuite) SetUpTest(c *C) { arvadostest.StartAPI() os.Setenv("ARVADOS_API_TOKEN", arvadostest.Dispatch1Token) + s.disp = Dispatcher{} + s.disp.setup() + s.slurm = slurmFake{} } -func (s *TestSuite) TearDownTest(c *C) { - os.Args = initialArgs +func (s *IntegrationSuite) TearDownTest(c *C) { arvadostest.ResetEnv() arvadostest.StopAPI() } -func (s *MockArvadosServerSuite) TearDownTest(c *C) { - arvadostest.ResetEnv() +type slurmFake struct { + didBatch [][]string + didCancel []string + didRelease []string + didRenice [][]string + queue string + // If non-nil, run this func during the 2nd+ call to Cancel() + onCancel func() + // Error returned by Batch() + errBatch error } -func (s *TestSuite) integrationTest(c *C, - newSqueueCmd func() *exec.Cmd, - newScancelCmd func(arvados.Container) *exec.Cmd, - newSbatchCmd func(arvados.Container) *exec.Cmd, - newScontrolCmd func(arvados.Container) *exec.Cmd, - sbatchCmdComps []string, - runContainer func(*dispatch.Dispatcher, arvados.Container)) arvados.Container { - arvadostest.ResetEnv() +func (sf *slurmFake) Batch(script io.Reader, args []string) error { + sf.didBatch = append(sf.didBatch, args) + return sf.errBatch +} - arv, err := arvadosclient.MakeArvadosClient() - c.Assert(err, IsNil) +func (sf *slurmFake) QueueCommand(args []string) *exec.Cmd { + return exec.Command("echo", sf.queue) +} - var sbatchCmdLine []string +func (sf *slurmFake) Release(name string) error { + sf.didRelease = append(sf.didRelease, name) + return nil +} - // Override sbatchCmd - defer func(orig func(arvados.Container) *exec.Cmd) { - sbatchCmd = orig - }(sbatchCmd) +func (sf *slurmFake) Renice(name string, nice int64) error { + sf.didRenice = append(sf.didRenice, []string{name, fmt.Sprintf("%d", nice)}) + return nil +} - if newSbatchCmd != nil { - sbatchCmd = newSbatchCmd - } else { - sbatchCmd = func(container arvados.Container) *exec.Cmd { - sbatchCmdLine = sbatchFunc(container).Args - return exec.Command("sh") - } +func (sf *slurmFake) Cancel(name string) error { + sf.didCancel = append(sf.didCancel, name) + if len(sf.didCancel) == 1 { + // simulate error on first attempt + return errors.New("something terrible happened") } + if sf.onCancel != nil { + sf.onCancel() + } + return nil +} - // Override squeueCmd - defer func(orig func() *exec.Cmd) { - squeueCmd = orig - }(squeueCmd) - squeueCmd = newSqueueCmd - - // Override scancel - defer func(orig func(arvados.Container) *exec.Cmd) { - scancelCmd = orig - }(scancelCmd) - scancelCmd = newScancelCmd +func (s *IntegrationSuite) integrationTest(c *C, + expectBatch [][]string, + runContainer func(*dispatch.Dispatcher, arvados.Container)) arvados.Container { + arvadostest.ResetEnv() - // Override scontrol - defer func(orig func(arvados.Container) *exec.Cmd) { - scontrolCmd = orig - }(scontrolCmd) - scontrolCmd = newScontrolCmd + arv, err := arvadosclient.MakeArvadosClient() + c.Assert(err, IsNil) // There should be one queued container params := arvadosclient.Dict{ @@ -119,33 +114,35 @@ func (s *TestSuite) integrationTest(c *C, c.Check(err, IsNil) c.Check(len(containers.Items), Equals, 1) - theConfig.CrunchRunCommand = []string{"echo"} + s.disp.CrunchRunCommand = []string{"echo"} ctx, cancel := context.WithCancel(context.Background()) doneRun := make(chan struct{}) - dispatcher := dispatch.Dispatcher{ + s.disp.Dispatcher = &dispatch.Dispatcher{ Arv: arv, - PollPeriod: time.Duration(1) * time.Second, + PollPeriod: time.Second, RunContainer: func(disp *dispatch.Dispatcher, ctr arvados.Container, status <-chan arvados.Container) { go func() { runContainer(disp, ctr) + s.slurm.queue = "" doneRun <- struct{}{} }() - run(disp, ctr, status) + s.disp.runContainer(disp, ctr, status) cancel() }, } - sqCheck = &SqueueChecker{Period: 500 * time.Millisecond} + s.disp.slurm = &s.slurm + s.disp.sqCheck = &SqueueChecker{Period: 500 * time.Millisecond, Slurm: s.disp.slurm} - err = dispatcher.Run(ctx) + err = s.disp.Dispatcher.Run(ctx) <-doneRun c.Assert(err, Equals, context.Canceled) - sqCheck.Stop() + s.disp.sqCheck.Stop() - c.Check(sbatchCmdLine, DeepEquals, sbatchCmdComps) + c.Check(s.slurm.didBatch, DeepEquals, expectBatch) // There should be no queued containers now err = arv.List("containers", params, &containers) @@ -159,78 +156,47 @@ func (s *TestSuite) integrationTest(c *C, return container } -func (s *TestSuite) TestIntegrationNormal(c *C) { - done := false +func (s *IntegrationSuite) TestNormal(c *C) { + s.slurm = slurmFake{queue: "zzzzz-dz642-queuedcontainer 10000 100 PENDING Resources\n"} container := s.integrationTest(c, - func() *exec.Cmd { - if done { - return exec.Command("true") - } else { - return exec.Command("echo", "zzzzz-dz642-queuedcontainer 9990 100") - } - }, - nil, nil, - nil, - []string(nil), func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) time.Sleep(3 * time.Second) dispatcher.UpdateState(container.UUID, dispatch.Complete) - done = true }) c.Check(container.State, Equals, arvados.ContainerStateComplete) } -func (s *TestSuite) TestIntegrationCancel(c *C) { - var cmd *exec.Cmd - var scancelCmdLine []string - attempt := 0 - +func (s *IntegrationSuite) TestCancel(c *C) { + s.slurm = slurmFake{queue: "zzzzz-dz642-queuedcontainer 10000 100 PENDING Resources\n"} + readyToCancel := make(chan bool) + s.slurm.onCancel = func() { <-readyToCancel } container := s.integrationTest(c, - func() *exec.Cmd { - if cmd != nil && cmd.ProcessState != nil { - return exec.Command("true") - } else { - return exec.Command("echo", "zzzzz-dz642-queuedcontainer 9990 100") - } - }, - func(container arvados.Container) *exec.Cmd { - if attempt++; attempt == 1 { - return exec.Command("false") - } else { - scancelCmdLine = scancelFunc(container).Args - cmd = exec.Command("echo") - return cmd - } - }, - nil, nil, - []string(nil), func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) - time.Sleep(1 * time.Second) + time.Sleep(time.Second) dispatcher.Arv.Update("containers", container.UUID, arvadosclient.Dict{ "container": arvadosclient.Dict{"priority": 0}}, nil) + readyToCancel <- true + close(readyToCancel) }) c.Check(container.State, Equals, arvados.ContainerStateCancelled) - c.Check(scancelCmdLine, DeepEquals, []string{"scancel", "--name=zzzzz-dz642-queuedcontainer"}) + c.Check(len(s.slurm.didCancel) > 1, Equals, true) + c.Check(s.slurm.didCancel[:2], DeepEquals, []string{"zzzzz-dz642-queuedcontainer", "zzzzz-dz642-queuedcontainer"}) } -func (s *TestSuite) TestIntegrationMissingFromSqueue(c *C) { +func (s *IntegrationSuite) TestMissingFromSqueue(c *C) { container := s.integrationTest(c, - func() *exec.Cmd { return exec.Command("echo") }, - nil, - nil, - nil, - []string{"sbatch", + [][]string{{ fmt.Sprintf("--job-name=%s", "zzzzz-dz642-queuedcontainer"), fmt.Sprintf("--mem=%d", 11445), fmt.Sprintf("--cpus-per-task=%d", 4), fmt.Sprintf("--tmp=%d", 45777), - fmt.Sprintf("--nice=%d", 9990)}, + fmt.Sprintf("--nice=%d", 10000)}}, func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) time.Sleep(3 * time.Second) @@ -239,15 +205,10 @@ func (s *TestSuite) TestIntegrationMissingFromSqueue(c *C) { c.Check(container.State, Equals, arvados.ContainerStateCancelled) } -func (s *TestSuite) TestSbatchFail(c *C) { +func (s *IntegrationSuite) TestSbatchFail(c *C) { + s.slurm = slurmFake{errBatch: errors.New("something terrible happened")} container := s.integrationTest(c, - func() *exec.Cmd { return exec.Command("echo") }, - nil, - func(container arvados.Container) *exec.Cmd { - return exec.Command("false") - }, - nil, - []string(nil), + [][]string{{"--job-name=zzzzz-dz642-queuedcontainer", "--mem=11445", "--cpus-per-task=4", "--tmp=45777", "--nice=10000"}}, func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) dispatcher.UpdateState(container.UUID, dispatch.Complete) @@ -262,18 +223,28 @@ func (s *TestSuite) TestSbatchFail(c *C) { {"object_uuid", "=", container.UUID}, {"event_type", "=", "dispatch"}, }}, &ll) + c.Assert(err, IsNil) c.Assert(len(ll.Items), Equals, 1) } -func (s *MockArvadosServerSuite) TestAPIErrorGettingContainers(c *C) { +type StubbedSuite struct { + disp Dispatcher +} + +func (s *StubbedSuite) SetUpTest(c *C) { + s.disp = Dispatcher{} + s.disp.setup() +} + +func (s *StubbedSuite) TestAPIErrorGettingContainers(c *C) { apiStubResponses := make(map[string]arvadostest.StubResponse) apiStubResponses["/arvados/v1/api_client_authorizations/current"] = arvadostest.StubResponse{200, `{"uuid":"` + arvadostest.Dispatch1AuthUUID + `"}`} apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)} - testWithServerStub(c, apiStubResponses, "echo", "Error getting list of containers") + s.testWithServerStub(c, apiStubResponses, "echo", "Error getting list of containers") } -func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) { +func (s *StubbedSuite) testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) { apiStub := arvadostest.ServerStub{apiStubResponses} api := httptest.NewServer(&apiStub) @@ -291,19 +262,19 @@ func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubRespon log.SetOutput(io.MultiWriter(buf, os.Stderr)) defer log.SetOutput(os.Stderr) - theConfig.CrunchRunCommand = []string{crunchCmd} + s.disp.CrunchRunCommand = []string{crunchCmd} ctx, cancel := context.WithCancel(context.Background()) dispatcher := dispatch.Dispatcher{ Arv: arv, - PollPeriod: time.Duration(1) * time.Second, + PollPeriod: time.Second, RunContainer: func(disp *dispatch.Dispatcher, ctr arvados.Container, status <-chan arvados.Container) { go func() { - time.Sleep(1 * time.Second) + time.Sleep(time.Second) disp.UpdateState(ctr.UUID, dispatch.Running) disp.UpdateState(ctr.UUID, dispatch.Complete) }() - run(disp, ctr, status) + s.disp.runContainer(disp, ctr, status) cancel() }, } @@ -321,15 +292,12 @@ func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubRespon c.Check(buf.String(), Matches, `(?ms).*`+expected+`.*`) } -func (s *MockArvadosServerSuite) TestNoSuchConfigFile(c *C) { - var config Config - err := readConfig(&config, "/nosuchdir89j7879/8hjwr7ojgyy7") +func (s *StubbedSuite) TestNoSuchConfigFile(c *C) { + err := s.disp.readConfig("/nosuchdir89j7879/8hjwr7ojgyy7") c.Assert(err, NotNil) } -func (s *MockArvadosServerSuite) TestBadSbatchArgsConfig(c *C) { - var config Config - +func (s *StubbedSuite) TestBadSbatchArgsConfig(c *C) { tmpfile, err := ioutil.TempFile(os.TempDir(), "config") c.Check(err, IsNil) defer os.Remove(tmpfile.Name()) @@ -337,13 +305,11 @@ func (s *MockArvadosServerSuite) TestBadSbatchArgsConfig(c *C) { _, err = tmpfile.Write([]byte(`{"SbatchArguments": "oops this is not a string array"}`)) c.Check(err, IsNil) - err = readConfig(&config, tmpfile.Name()) + err = s.disp.readConfig(tmpfile.Name()) c.Assert(err, NotNil) } -func (s *MockArvadosServerSuite) TestNoSuchArgInConfigIgnored(c *C) { - var config Config - +func (s *StubbedSuite) TestNoSuchArgInConfigIgnored(c *C) { tmpfile, err := ioutil.TempFile(os.TempDir(), "config") c.Check(err, IsNil) defer os.Remove(tmpfile.Name()) @@ -351,14 +317,12 @@ func (s *MockArvadosServerSuite) TestNoSuchArgInConfigIgnored(c *C) { _, err = tmpfile.Write([]byte(`{"NoSuchArg": "Nobody loves me, not one tiny hunk."}`)) c.Check(err, IsNil) - err = readConfig(&config, tmpfile.Name()) + err = s.disp.readConfig(tmpfile.Name()) c.Assert(err, IsNil) - c.Check(0, Equals, len(config.SbatchArguments)) + c.Check(0, Equals, len(s.disp.SbatchArguments)) } -func (s *MockArvadosServerSuite) TestReadConfig(c *C) { - var config Config - +func (s *StubbedSuite) TestReadConfig(c *C) { tmpfile, err := ioutil.TempFile(os.TempDir(), "config") c.Check(err, IsNil) defer os.Remove(tmpfile.Name()) @@ -368,90 +332,90 @@ func (s *MockArvadosServerSuite) TestReadConfig(c *C) { _, err = tmpfile.Write([]byte(argsS)) c.Check(err, IsNil) - err = readConfig(&config, tmpfile.Name()) + err = s.disp.readConfig(tmpfile.Name()) c.Assert(err, IsNil) - c.Check(3, Equals, len(config.SbatchArguments)) - c.Check(args, DeepEquals, config.SbatchArguments) -} - -func (s *MockArvadosServerSuite) TestSbatchFuncWithNoConfigArgs(c *C) { - testSbatchFuncWithArgs(c, nil) + c.Check(args, DeepEquals, s.disp.SbatchArguments) } -func (s *MockArvadosServerSuite) TestSbatchFuncWithEmptyConfigArgs(c *C) { - testSbatchFuncWithArgs(c, []string{}) -} +func (s *StubbedSuite) TestSbatchArgs(c *C) { + container := arvados.Container{ + UUID: "123", + RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 2}, + Priority: 1, + } -func (s *MockArvadosServerSuite) TestSbatchFuncWithConfigArgs(c *C) { - testSbatchFuncWithArgs(c, []string{"--arg1=v1", "--arg2"}) + for _, defaults := range [][]string{ + nil, + {}, + {"--arg1=v1", "--arg2"}, + } { + c.Logf("%#v", defaults) + s.disp.SbatchArguments = defaults + + args, err := s.disp.sbatchArgs(container) + c.Check(args, DeepEquals, append(defaults, "--job-name=123", "--mem=239", "--cpus-per-task=2", "--tmp=0", "--nice=10000")) + c.Check(err, IsNil) + } } -func testSbatchFuncWithArgs(c *C, args []string) { - theConfig.SbatchArguments = append(theConfig.SbatchArguments, args...) - +func (s *StubbedSuite) TestSbatchInstanceTypeConstraint(c *C) { container := arvados.Container{ UUID: "123", RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 2}, - Priority: 1} - sbatchCmd := sbatchFunc(container) - - var expected []string - expected = append(expected, "sbatch") - expected = append(expected, theConfig.SbatchArguments...) - expected = append(expected, "--job-name=123", "--mem=239", "--cpus-per-task=2", "--tmp=0", "--nice=9990") + Priority: 1, + } - c.Check(sbatchCmd.Args, DeepEquals, expected) + for _, trial := range []struct { + types []arvados.InstanceType + sbatchArgs []string + err error + }{ + // Choose node type => use --constraint arg + { + types: []arvados.InstanceType{ + {Name: "a1.tiny", Price: 0.02, RAM: 128000000, VCPUs: 1}, + {Name: "a1.small", Price: 0.04, RAM: 256000000, VCPUs: 2}, + {Name: "a1.medium", Price: 0.08, RAM: 512000000, VCPUs: 4}, + {Name: "a1.large", Price: 0.16, RAM: 1024000000, VCPUs: 8}, + }, + sbatchArgs: []string{"--constraint=instancetype=a1.medium"}, + }, + // No node types configured => no slurm constraint + { + types: nil, + sbatchArgs: nil, + }, + // No node type is big enough => error + { + types: []arvados.InstanceType{ + {Name: "a1.tiny", Price: 0.02, RAM: 128000000, VCPUs: 1}, + }, + err: dispatchcloud.ErrConstraintsNotSatisfiable, + }, + } { + c.Logf("%#v", trial) + s.disp.cluster = &arvados.Cluster{InstanceTypes: trial.types} + + args, err := s.disp.sbatchArgs(container) + c.Check(err, Equals, trial.err) + if trial.err == nil { + c.Check(args, DeepEquals, append([]string{"--job-name=123", "--mem=239", "--cpus-per-task=2", "--tmp=0", "--nice=10000"}, trial.sbatchArgs...)) + } + } } -func (s *MockArvadosServerSuite) TestSbatchPartition(c *C) { - theConfig.SbatchArguments = nil +func (s *StubbedSuite) TestSbatchPartition(c *C) { container := arvados.Container{ UUID: "123", RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 1}, SchedulingParameters: arvados.SchedulingParameters{Partitions: []string{"blurb", "b2"}}, - Priority: 1} - sbatchCmd := sbatchFunc(container) - - var expected []string - expected = append(expected, "sbatch") - expected = append(expected, "--job-name=123", "--mem=239", "--cpus-per-task=1", "--tmp=0", "--nice=9990", "--partition=blurb,b2") - - c.Check(sbatchCmd.Args, DeepEquals, expected) -} - -func (s *TestSuite) TestIntegrationChangePriority(c *C) { - var scontrolCmdLine []string - step := 0 + Priority: 1, + } - container := s.integrationTest(c, - func() *exec.Cmd { - if step == 0 { - return exec.Command("echo", "zzzzz-dz642-queuedcontainer 9990 100") - } else if step == 1 { - return exec.Command("echo", "zzzzz-dz642-queuedcontainer 4000 100") - } else { - return exec.Command("echo") - } - }, - func(arvados.Container) *exec.Cmd { return exec.Command("true") }, - nil, - func(container arvados.Container) *exec.Cmd { - scontrolCmdLine = scontrolFunc(container).Args - step = 1 - return exec.Command("true") - }, - []string(nil), - func(dispatcher *dispatch.Dispatcher, container arvados.Container) { - dispatcher.UpdateState(container.UUID, dispatch.Running) - time.Sleep(1 * time.Second) - dispatcher.Arv.Update("containers", container.UUID, - arvadosclient.Dict{ - "container": arvadosclient.Dict{"priority": 600}}, - nil) - time.Sleep(1 * time.Second) - step = 2 - dispatcher.UpdateState(container.UUID, dispatch.Complete) - }) - c.Check(container.State, Equals, arvados.ContainerStateComplete) - c.Check(scontrolCmdLine, DeepEquals, []string{"scontrol", "update", "JobName=zzzzz-dz642-queuedcontainer", "Nice=4000"}) + args, err := s.disp.sbatchArgs(container) + c.Check(args, DeepEquals, []string{ + "--job-name=123", "--mem=239", "--cpus-per-task=1", "--tmp=0", "--nice=10000", + "--partition=blurb,b2", + }) + c.Check(err, IsNil) }