X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0f644e242ef37c911ad3dc25aca8135c339de349..47e59a35d5ed9b2cdb052894d741972324058505:/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 5ab0e4fcb3..a823755379 100644 --- a/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go +++ b/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go @@ -64,6 +64,101 @@ func (s *MockArvadosServerSuite) TearDownTest(c *C) { arvadostest.ResetEnv() } +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() + + arv, err := arvadosclient.MakeArvadosClient() + c.Assert(err, IsNil) + + var sbatchCmdLine []string + + // Override sbatchCmd + defer func(orig func(arvados.Container) *exec.Cmd) { + sbatchCmd = orig + }(sbatchCmd) + + if newSbatchCmd != nil { + sbatchCmd = newSbatchCmd + } else { + 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 + + // Override scancel + defer func(orig func(arvados.Container) *exec.Cmd) { + scancelCmd = orig + }(scancelCmd) + scancelCmd = newScancelCmd + + // Override scontrol + defer func(orig func(arvados.Container) *exec.Cmd) { + scontrolCmd = orig + }(scontrolCmd) + scontrolCmd = newScontrolCmd + + // There should be one queued container + params := arvadosclient.Dict{ + "filters": [][]string{{"state", "=", "Queued"}}, + } + var containers arvados.ContainerList + err = arv.List("containers", params, &containers) + c.Check(err, IsNil) + c.Check(len(containers.Items), Equals, 1) + + theConfig.CrunchRunCommand = []string{"echo"} + + ctx, cancel := context.WithCancel(context.Background()) + doneRun := make(chan struct{}) + + dispatcher := dispatch.Dispatcher{ + Arv: arv, + PollPeriod: time.Duration(1) * time.Second, + RunContainer: func(disp *dispatch.Dispatcher, ctr arvados.Container, status <-chan arvados.Container) { + go func() { + runContainer(disp, ctr) + doneRun <- struct{}{} + }() + run(disp, ctr, status) + cancel() + }, + } + + sqCheck = &SqueueChecker{Period: 500 * time.Millisecond} + + err = dispatcher.Run(ctx) + <-doneRun + c.Assert(err, Equals, context.Canceled) + + sqCheck.Stop() + + c.Check(sbatchCmdLine, DeepEquals, sbatchCmdComps) + + // There should be no queued containers now + err = arv.List("containers", params, &containers) + c.Check(err, IsNil) + c.Check(len(containers.Items), Equals, 0) + + // Previously "Queued" container should now be in "Complete" state + var container arvados.Container + err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container) + c.Check(err, IsNil) + return container +} + func (s *TestSuite) TestIntegrationNormal(c *C) { done := false container := s.integrationTest(c, @@ -71,11 +166,12 @@ func (s *TestSuite) TestIntegrationNormal(c *C) { if done { return exec.Command("true") } else { - return exec.Command("echo", "zzzzz-dz642-queuedcontainer") + 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) @@ -96,7 +192,7 @@ func (s *TestSuite) TestIntegrationCancel(c *C) { if cmd != nil && cmd.ProcessState != nil { return exec.Command("true") } else { - return exec.Command("echo", "zzzzz-dz642-queuedcontainer") + return exec.Command("echo", "zzzzz-dz642-queuedcontainer 9990 100") } }, func(container arvados.Container) *exec.Cmd { @@ -109,6 +205,7 @@ func (s *TestSuite) TestIntegrationCancel(c *C) { } }, nil, + nil, []string(nil), func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) @@ -127,11 +224,13 @@ func (s *TestSuite) TestIntegrationMissingFromSqueue(c *C) { func() *exec.Cmd { return exec.Command("echo") }, nil, nil, + nil, []string{"sbatch", 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("--tmp=%d", 45777), + fmt.Sprintf("--nice=%d", 9990)}, func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) time.Sleep(3 * time.Second) @@ -147,6 +246,7 @@ func (s *TestSuite) TestSbatchFail(c *C) { func(container arvados.Container) *exec.Cmd { return exec.Command("false") }, + nil, []string(nil), func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) @@ -159,100 +259,12 @@ func (s *TestSuite) TestSbatchFail(c *C) { var ll arvados.LogList err = arv.List("logs", arvadosclient.Dict{"filters": [][]string{ - []string{"object_uuid", "=", container.UUID}, - []string{"event_type", "=", "dispatch"}, + {"object_uuid", "=", container.UUID}, + {"event_type", "=", "dispatch"}, }}, &ll) c.Assert(len(ll.Items), Equals, 1) } -func (s *TestSuite) integrationTest(c *C, - newSqueueCmd func() *exec.Cmd, - newScancelCmd func(arvados.Container) *exec.Cmd, - newSbatchCmd func(arvados.Container) *exec.Cmd, - sbatchCmdComps []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) - - if newSbatchCmd != nil { - sbatchCmd = newSbatchCmd - } else { - 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 - - // Override scancel - defer func(orig func(arvados.Container) *exec.Cmd) { - scancelCmd = orig - }(scancelCmd) - scancelCmd = newScancelCmd - - // There should be one queued container - params := arvadosclient.Dict{ - "filters": [][]string{{"state", "=", "Queued"}}, - } - var containers arvados.ContainerList - err = arv.List("containers", params, &containers) - c.Check(err, IsNil) - c.Check(len(containers.Items), Equals, 1) - - theConfig.CrunchRunCommand = []string{"echo"} - - ctx, cancel := context.WithCancel(context.Background()) - doneRun := make(chan struct{}) - - dispatcher := dispatch.Dispatcher{ - Arv: arv, - PollPeriod: time.Duration(1) * time.Second, - RunContainer: func(disp *dispatch.Dispatcher, ctr arvados.Container, status <-chan arvados.Container) { - go func() { - runContainer(disp, ctr) - doneRun <- struct{}{} - }() - run(disp, ctr, status) - cancel() - }, - } - - sqCheck = &SqueueChecker{Period: 500 * time.Millisecond} - - err = dispatcher.Run(ctx) - <-doneRun - c.Assert(err, Equals, context.Canceled) - - sqCheck.Stop() - - c.Check(sbatchCmdLine, DeepEquals, sbatchCmdComps) - - // There should be no queued containers now - err = arv.List("containers", params, &containers) - c.Check(err, IsNil) - c.Check(len(containers.Items), Equals, 0) - - // Previously "Queued" container should now be in "Complete" state - var container arvados.Container - err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container) - c.Check(err, IsNil) - return container -} - func (s *MockArvadosServerSuite) TestAPIErrorGettingContainers(c *C) { apiStubResponses := make(map[string]arvadostest.StubResponse) apiStubResponses["/arvados/v1/api_client_authorizations/current"] = arvadostest.StubResponse{200, `{"uuid":"` + arvadostest.Dispatch1AuthUUID + `"}`} @@ -377,25 +389,69 @@ func (s *MockArvadosServerSuite) TestSbatchFuncWithConfigArgs(c *C) { func testSbatchFuncWithArgs(c *C, args []string) { theConfig.SbatchArguments = append(theConfig.SbatchArguments, args...) - container := arvados.Container{UUID: "123", RuntimeConstraints: arvados.RuntimeConstraints{RAM: 250000000, VCPUs: 2}} + 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") + expected = append(expected, "--job-name=123", "--mem=239", "--cpus-per-task=2", "--tmp=0", "--nice=9990") c.Check(sbatchCmd.Args, DeepEquals, expected) } 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"}}} + 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", "--partition=blurb,b2") + 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 + + 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"}) +}