X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/4f4f6f5fe9367d5cd6c57070fd8a223efc87cb21..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 fbea48e548..499f6d5d72 100644 --- a/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go +++ b/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go @@ -1,12 +1,14 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( "bytes" + "context" + "errors" "fmt" - "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/dispatch" "io" "io/ioutil" "log" @@ -18,6 +20,11 @@ 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" + "git.curoverse.com/arvados.git/sdk/go/dispatch" . "gopkg.in/check.v1" ) @@ -26,115 +33,79 @@ func Test(t *testing.T) { TestingT(t) } -var _ = Suite(&TestSuite{}) -var _ = Suite(&MockArvadosServerSuite{}) - -type TestSuite struct{} -type MockArvadosServerSuite struct{} +var _ = Suite(&IntegrationSuite{}) +var _ = Suite(&StubbedSuite{}) -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) TestIntegrationNormal(c *C) { - container := s.integrationTest(c, func() *exec.Cmd { return exec.Command("echo", "zzzzz-dz642-queuedcontainer") }, - []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) - }) - c.Check(container.State, Equals, arvados.ContainerStateComplete) +func (sf *slurmFake) Batch(script io.Reader, args []string) error { + sf.didBatch = append(sf.didBatch, args) + return sf.errBatch } -func (s *TestSuite) TestIntegrationCancel(c *C) { +func (sf *slurmFake) QueueCommand(args []string) *exec.Cmd { + return exec.Command("echo", sf.queue) +} - // Override sbatchCmd - var scancelCmdLine []string - defer func(orig func(arvados.Container) *exec.Cmd) { - scancelCmd = orig - }(scancelCmd) - scancelCmd = func(container arvados.Container) *exec.Cmd { - scancelCmdLine = scancelFunc(container).Args - return exec.Command("echo") - } +func (sf *slurmFake) Release(name string) error { + sf.didRelease = append(sf.didRelease, name) + return nil +} - container := s.integrationTest(c, func() *exec.Cmd { return exec.Command("echo", "zzzzz-dz642-queuedcontainer") }, - []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": 0}}, - nil) - }) - c.Check(container.State, Equals, arvados.ContainerStateCancelled) - c.Check(scancelCmdLine, DeepEquals, []string{"scancel", "--name=zzzzz-dz642-queuedcontainer"}) +func (sf *slurmFake) Renice(name string, nice int64) error { + sf.didRenice = append(sf.didRenice, []string{name, fmt.Sprintf("%d", nice)}) + return nil } -func (s *TestSuite) TestIntegrationMissingFromSqueue(c *C) { - container := s.integrationTest(c, func() *exec.Cmd { return exec.Command("echo") }, []string{"sbatch", "--share", - fmt.Sprintf("--job-name=%s", "zzzzz-dz642-queuedcontainer"), - fmt.Sprintf("--mem-per-cpu=%d", 2862), - fmt.Sprintf("--cpus-per-task=%d", 4)}, - func(dispatcher *dispatch.Dispatcher, container arvados.Container) { - dispatcher.UpdateState(container.UUID, dispatch.Running) - time.Sleep(3 * time.Second) - dispatcher.UpdateState(container.UUID, dispatch.Complete) - }) - c.Check(container.State, Equals, arvados.ContainerStateCancelled) +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 } -func (s *TestSuite) integrationTest(c *C, - newSqueueCmd func() *exec.Cmd, - sbatchCmdComps []string, +func (s *IntegrationSuite) integrationTest(c *C, + expectBatch [][]string, runContainer func(*dispatch.Dispatcher, arvados.Container)) arvados.Container { arvadostest.ResetEnv() arv, err := arvadosclient.MakeArvadosClient() c.Assert(err, IsNil) - var sbatchCmdLine []string - - // Override sbatchCmd - defer func(orig func(arvados.Container) *exec.Cmd) { - sbatchCmd = orig - }(sbatchCmd) - sbatchCmd = func(container arvados.Container) *exec.Cmd { - sbatchCmdLine = sbatchFunc(container).Args - return exec.Command("sh") - } - - // Override squeueCmd - defer func(orig func() *exec.Cmd) { - squeueCmd = orig - }(squeueCmd) - squeueCmd = newSqueueCmd - - // There should be no queued containers now + // There should be one queued container params := arvadosclient.Dict{ "filters": [][]string{{"state", "=", "Queued"}}, } @@ -143,29 +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"} - doneProcessing := make(chan struct{}) - dispatcher := dispatch.Dispatcher{ - Arv: arv, - PollInterval: time.Duration(1) * time.Second, - RunContainer: func(dispatcher *dispatch.Dispatcher, - container arvados.Container, - status chan arvados.Container) { - go runContainer(dispatcher, container) - run(dispatcher, container, status) - doneProcessing <- struct{}{} + ctx, cancel := context.WithCancel(context.Background()) + doneRun := make(chan struct{}) + + s.disp.Dispatcher = &dispatch.Dispatcher{ + Arv: arv, + 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{}{} + }() + s.disp.runContainer(disp, ctr, status) + cancel() }, - DoneProcessing: doneProcessing} + } - squeueUpdater.StartMonitor(time.Duration(500) * time.Millisecond) + s.disp.slurm = &s.slurm + s.disp.sqCheck = &SqueueChecker{Period: 500 * time.Millisecond, Slurm: s.disp.slurm} - err = dispatcher.RunDispatcher() - c.Assert(err, IsNil) + err = s.disp.Dispatcher.Run(ctx) + <-doneRun + c.Assert(err, Equals, context.Canceled) - squeueUpdater.Done() + 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) @@ -179,15 +156,95 @@ func (s *TestSuite) integrationTest(c *C, return container } -func (s *MockArvadosServerSuite) TestAPIErrorGettingContainers(c *C) { +func (s *IntegrationSuite) TestNormal(c *C) { + s.slurm = slurmFake{queue: "zzzzz-dz642-queuedcontainer 10000 100 PENDING Resources\n"} + container := s.integrationTest(c, + 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) + }) + c.Check(container.State, Equals, arvados.ContainerStateComplete) +} + +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, + nil, + func(dispatcher *dispatch.Dispatcher, container arvados.Container) { + dispatcher.UpdateState(container.UUID, dispatch.Running) + 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(len(s.slurm.didCancel) > 1, Equals, true) + c.Check(s.slurm.didCancel[:2], DeepEquals, []string{"zzzzz-dz642-queuedcontainer", "zzzzz-dz642-queuedcontainer"}) +} + +func (s *IntegrationSuite) TestMissingFromSqueue(c *C) { + container := s.integrationTest(c, + [][]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", 10000)}}, + func(dispatcher *dispatch.Dispatcher, container arvados.Container) { + dispatcher.UpdateState(container.UUID, dispatch.Running) + time.Sleep(3 * time.Second) + dispatcher.UpdateState(container.UUID, dispatch.Complete) + }) + c.Check(container.State, Equals, arvados.ContainerStateCancelled) +} + +func (s *IntegrationSuite) TestSbatchFail(c *C) { + s.slurm = slurmFake{errBatch: errors.New("something terrible happened")} + container := s.integrationTest(c, + [][]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) + }) + c.Check(container.State, Equals, arvados.ContainerStateComplete) + + arv, err := arvadosclient.MakeArvadosClient() + c.Assert(err, IsNil) + + var ll arvados.LogList + err = arv.List("logs", arvadosclient.Dict{"filters": [][]string{ + {"object_uuid", "=", container.UUID}, + {"event_type", "=", "dispatch"}, + }}, &ll) + c.Assert(err, IsNil) + c.Assert(len(ll.Items), Equals, 1) +} + +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) @@ -205,47 +262,42 @@ 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} - doneProcessing := make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) dispatcher := dispatch.Dispatcher{ - Arv: arv, - PollInterval: time.Duration(1) * time.Second, - RunContainer: func(dispatcher *dispatch.Dispatcher, - container arvados.Container, - status chan arvados.Container) { + Arv: arv, + PollPeriod: time.Second, + RunContainer: func(disp *dispatch.Dispatcher, ctr arvados.Container, status <-chan arvados.Container) { go func() { - time.Sleep(1 * time.Second) - dispatcher.UpdateState(container.UUID, dispatch.Running) - dispatcher.UpdateState(container.UUID, dispatch.Complete) + time.Sleep(time.Second) + disp.UpdateState(ctr.UUID, dispatch.Running) + disp.UpdateState(ctr.UUID, dispatch.Complete) }() - run(dispatcher, container, status) - doneProcessing <- struct{}{} + s.disp.runContainer(disp, ctr, status) + cancel() }, - DoneProcessing: doneProcessing} + } go func() { for i := 0; i < 80 && !strings.Contains(buf.String(), expected); i++ { time.Sleep(100 * time.Millisecond) } - dispatcher.DoneProcessing <- struct{}{} + cancel() }() - err := dispatcher.RunDispatcher() - c.Assert(err, IsNil) + err := dispatcher.Run(ctx) + c.Assert(err, Equals, context.Canceled) 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()) @@ -253,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()) @@ -267,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()) @@ -284,46 +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...) - - container := arvados.Container{UUID: "123", RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 2}} - sbatchCmd := sbatchFunc(container) - - var expected []string - expected = append(expected, "sbatch", "--share") - expected = append(expected, theConfig.SbatchArguments...) - expected = append(expected, "--job-name=123", "--mem-per-cpu=120", "--cpus-per-task=2") +func (s *StubbedSuite) TestSbatchInstanceTypeConstraint(c *C) { + container := arvados.Container{ + UUID: "123", + RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 2}, + 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 - container := arvados.Container{UUID: "123", RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 1}, SchedulingParameters: arvados.SchedulingParameters{Partitions: []string{"blurb", "b2"}}} - sbatchCmd := sbatchFunc(container) - - var expected []string - expected = append(expected, "sbatch", "--share") - expected = append(expected, "--job-name=123", "--mem-per-cpu=239", "--cpus-per-task=1", "--partition=blurb,b2") +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, + } - c.Check(sbatchCmd.Args, DeepEquals, expected) + 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) }