X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/10f08c358c12468119dc2621c48b68d6d33417da..355173ba2e8c42b29011493d1d8c7cc4d69295c6:/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 459b7c6ec2..23a8a0ca01 100644 --- a/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go +++ b/services/crunch-dispatch-slurm/crunch-dispatch-slurm_test.go @@ -20,11 +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" - "git.curoverse.com/arvados.git/services/dispatchcloud" . "gopkg.in/check.v1" ) @@ -37,12 +37,16 @@ var _ = Suite(&IntegrationSuite{}) var _ = Suite(&StubbedSuite{}) type IntegrationSuite struct { - cmd command + disp Dispatcher + slurm slurmFake } 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 *IntegrationSuite) TearDownTest(c *C) { @@ -51,10 +55,11 @@ func (s *IntegrationSuite) TearDownTest(c *C) { } type slurmFake struct { - didBatch [][]string - didCancel []string - didRenice [][]string - queue string + 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() @@ -70,7 +75,12 @@ func (sf *slurmFake) QueueCommand(args []string) *exec.Cmd { return exec.Command("echo", sf.queue) } -func (sf *slurmFake) Renice(name string, nice int) error { +func (sf *slurmFake) Release(name string) error { + sf.didRelease = append(sf.didRelease, name) + return nil +} + +func (sf *slurmFake) Renice(name string, nice int64) error { sf.didRenice = append(sf.didRenice, []string{name, fmt.Sprintf("%d", nice)}) return nil } @@ -87,7 +97,7 @@ func (sf *slurmFake) Cancel(name string) error { return nil } -func (s *IntegrationSuite) integrationTest(c *C, slurm *slurmFake, +func (s *IntegrationSuite) integrationTest(c *C, expectBatch [][]string, runContainer func(*dispatch.Dispatcher, arvados.Container)) arvados.Container { arvadostest.ResetEnv() @@ -95,8 +105,6 @@ func (s *IntegrationSuite) integrationTest(c *C, slurm *slurmFake, arv, err := arvadosclient.MakeArvadosClient() c.Assert(err, IsNil) - s.cmd.slurm = slurm - // There should be one queued container params := arvadosclient.Dict{ "filters": [][]string{{"state", "=", "Queued"}}, @@ -106,34 +114,35 @@ func (s *IntegrationSuite) integrationTest(c *C, slurm *slurmFake, c.Check(err, IsNil) c.Check(len(containers.Items), Equals, 1) - s.cmd.CrunchRunCommand = []string{"echo"} + s.disp.CrunchRunCommand = []string{"echo"} ctx, cancel := context.WithCancel(context.Background()) doneRun := make(chan struct{}) - s.cmd.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) - slurm.queue = "" + s.slurm.queue = "" doneRun <- struct{}{} }() - s.cmd.run(disp, ctr, status) + s.disp.runContainer(disp, ctr, status) cancel() }, } - s.cmd.sqCheck = &SqueueChecker{Period: 500 * time.Millisecond, Slurm: slurm} + s.disp.slurm = &s.slurm + s.disp.sqCheck = &SqueueChecker{Period: 500 * time.Millisecond, Slurm: s.disp.slurm} - err = s.cmd.dispatcher.Run(ctx) + err = s.disp.Dispatcher.Run(ctx) <-doneRun c.Assert(err, Equals, context.Canceled) - s.cmd.sqCheck.Stop() + s.disp.sqCheck.Stop() - c.Check(slurm.didBatch, DeepEquals, expectBatch) + c.Check(s.slurm.didBatch, DeepEquals, expectBatch) // There should be no queued containers now err = arv.List("containers", params, &containers) @@ -148,8 +157,8 @@ func (s *IntegrationSuite) integrationTest(c *C, slurm *slurmFake, } func (s *IntegrationSuite) TestNormal(c *C) { + s.slurm = slurmFake{queue: "zzzzz-dz642-queuedcontainer 10000 100 PENDING Resources\n"} container := s.integrationTest(c, - &slurmFake{queue: "zzzzz-dz642-queuedcontainer 9990 100\n"}, nil, func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) @@ -160,11 +169,10 @@ func (s *IntegrationSuite) TestNormal(c *C) { } func (s *IntegrationSuite) TestCancel(c *C) { - slurm := &slurmFake{queue: "zzzzz-dz642-queuedcontainer 9990 100\n"} + s.slurm = slurmFake{queue: "zzzzz-dz642-queuedcontainer 10000 100 PENDING Resources\n"} readyToCancel := make(chan bool) - slurm.onCancel = func() { <-readyToCancel } + s.slurm.onCancel = func() { <-readyToCancel } container := s.integrationTest(c, - slurm, nil, func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) @@ -177,18 +185,19 @@ func (s *IntegrationSuite) TestCancel(c *C) { close(readyToCancel) }) c.Check(container.State, Equals, arvados.ContainerStateCancelled) - c.Check(len(slurm.didCancel) > 1, Equals, true) - c.Check(slurm.didCancel[:2], DeepEquals, []string{"zzzzz-dz642-queuedcontainer", "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 *IntegrationSuite) TestMissingFromSqueue(c *C) { - container := s.integrationTest(c, &slurmFake{}, + container := s.integrationTest(c, [][]string{{ fmt.Sprintf("--job-name=%s", "zzzzz-dz642-queuedcontainer"), + fmt.Sprintf("--nice=%d", 10000), fmt.Sprintf("--mem=%d", 11445), fmt.Sprintf("--cpus-per-task=%d", 4), 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) @@ -198,9 +207,9 @@ func (s *IntegrationSuite) TestMissingFromSqueue(c *C) { } func (s *IntegrationSuite) TestSbatchFail(c *C) { + s.slurm = slurmFake{errBatch: errors.New("something terrible happened")} container := s.integrationTest(c, - &slurmFake{errBatch: errors.New("something terrible happened")}, - [][]string{{"--job-name=zzzzz-dz642-queuedcontainer", "--mem=11445", "--cpus-per-task=4", "--tmp=45777", "--nice=9990"}}, + [][]string{{"--job-name=zzzzz-dz642-queuedcontainer", "--nice=10000", "--mem=11445", "--cpus-per-task=4", "--tmp=45777"}}, func(dispatcher *dispatch.Dispatcher, container arvados.Container) { dispatcher.UpdateState(container.UUID, dispatch.Running) dispatcher.UpdateState(container.UUID, dispatch.Complete) @@ -215,33 +224,17 @@ func (s *IntegrationSuite) TestSbatchFail(c *C) { {"object_uuid", "=", container.UUID}, {"event_type", "=", "dispatch"}, }}, &ll) + c.Assert(err, IsNil) c.Assert(len(ll.Items), Equals, 1) } -func (s *IntegrationSuite) TestChangePriority(c *C) { - slurm := &slurmFake{queue: "zzzzz-dz642-queuedcontainer 9990 100\n"} - container := s.integrationTest(c, slurm, 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": 600}}, - nil) - time.Sleep(time.Second) - dispatcher.UpdateState(container.UUID, dispatch.Complete) - }) - c.Check(container.State, Equals, arvados.ContainerStateComplete) - c.Assert(len(slurm.didRenice), Not(Equals), 0) - c.Check(slurm.didRenice[len(slurm.didRenice)-1], DeepEquals, []string{"zzzzz-dz642-queuedcontainer", "4000"}) -} - type StubbedSuite struct { - cmd command + disp Dispatcher } func (s *StubbedSuite) SetUpTest(c *C) { - s.cmd = command{} + s.disp = Dispatcher{} + s.disp.setup() } func (s *StubbedSuite) TestAPIErrorGettingContainers(c *C) { @@ -270,19 +263,19 @@ func (s *StubbedSuite) testWithServerStub(c *C, apiStubResponses map[string]arva log.SetOutput(io.MultiWriter(buf, os.Stderr)) defer log.SetOutput(os.Stderr) - s.cmd.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) }() - s.cmd.run(disp, ctr, status) + s.disp.runContainer(disp, ctr, status) cancel() }, } @@ -301,7 +294,7 @@ func (s *StubbedSuite) testWithServerStub(c *C, apiStubResponses map[string]arva } func (s *StubbedSuite) TestNoSuchConfigFile(c *C) { - err := s.cmd.readConfig("/nosuchdir89j7879/8hjwr7ojgyy7") + err := s.disp.readConfig("/nosuchdir89j7879/8hjwr7ojgyy7") c.Assert(err, NotNil) } @@ -313,7 +306,7 @@ func (s *StubbedSuite) TestBadSbatchArgsConfig(c *C) { _, err = tmpfile.Write([]byte(`{"SbatchArguments": "oops this is not a string array"}`)) c.Check(err, IsNil) - err = s.cmd.readConfig(tmpfile.Name()) + err = s.disp.readConfig(tmpfile.Name()) c.Assert(err, NotNil) } @@ -325,9 +318,9 @@ func (s *StubbedSuite) TestNoSuchArgInConfigIgnored(c *C) { _, err = tmpfile.Write([]byte(`{"NoSuchArg": "Nobody loves me, not one tiny hunk."}`)) c.Check(err, IsNil) - err = s.cmd.readConfig(tmpfile.Name()) + err = s.disp.readConfig(tmpfile.Name()) c.Assert(err, IsNil) - c.Check(0, Equals, len(s.cmd.SbatchArguments)) + c.Check(0, Equals, len(s.disp.SbatchArguments)) } func (s *StubbedSuite) TestReadConfig(c *C) { @@ -340,9 +333,9 @@ func (s *StubbedSuite) TestReadConfig(c *C) { _, err = tmpfile.Write([]byte(argsS)) c.Check(err, IsNil) - err = s.cmd.readConfig(tmpfile.Name()) + err = s.disp.readConfig(tmpfile.Name()) c.Assert(err, IsNil) - c.Check(args, DeepEquals, s.cmd.SbatchArguments) + c.Check(args, DeepEquals, s.disp.SbatchArguments) } func (s *StubbedSuite) TestSbatchArgs(c *C) { @@ -358,10 +351,10 @@ func (s *StubbedSuite) TestSbatchArgs(c *C) { {"--arg1=v1", "--arg2"}, } { c.Logf("%#v", defaults) - s.cmd.SbatchArguments = defaults + s.disp.SbatchArguments = defaults - args, err := s.cmd.sbatchArgs(container) - c.Check(args, DeepEquals, append(defaults, "--job-name=123", "--mem=239", "--cpus-per-task=2", "--tmp=0", "--nice=9990")) + args, err := s.disp.sbatchArgs(container) + c.Check(args, DeepEquals, append(defaults, "--job-name=123", "--nice=10000", "--mem=239", "--cpus-per-task=2", "--tmp=0")) c.Check(err, IsNil) } } @@ -374,39 +367,42 @@ func (s *StubbedSuite) TestSbatchInstanceTypeConstraint(c *C) { } for _, trial := range []struct { - types []arvados.InstanceType + types map[string]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}, + types: map[string]arvados.InstanceType{ + "a1.tiny": {Name: "a1.tiny", Price: 0.02, RAM: 128000000, VCPUs: 1}, + "a1.small": {Name: "a1.small", Price: 0.04, RAM: 256000000, VCPUs: 2}, + "a1.medium": {Name: "a1.medium", Price: 0.08, RAM: 512000000, VCPUs: 4}, + "a1.large": {Name: "a1.large", Price: 0.16, RAM: 1024000000, VCPUs: 8}, }, - sbatchArgs: []string{"--constraint=a1.small"}, + sbatchArgs: []string{"--constraint=instancetype=a1.medium"}, }, // No node types configured => no slurm constraint { types: nil, - sbatchArgs: nil, + sbatchArgs: []string{"--mem=239", "--cpus-per-task=2", "--tmp=0"}, }, // No node type is big enough => error { - types: []arvados.InstanceType{ - {Name: "a1.tiny", Price: 0.02, RAM: 128000000, VCPUs: 1}, + types: map[string]arvados.InstanceType{ + "a1.tiny": {Name: "a1.tiny", Price: 0.02, RAM: 128000000, VCPUs: 1}, }, - err: dispatchcloud.ErrConstraintsNotSatisfiable, + err: dispatchcloud.ConstraintsNotSatisfiableError{}, }, } { c.Logf("%#v", trial) - s.cmd.cluster = &arvados.Cluster{InstanceTypes: trial.types} + s.disp.cluster = &arvados.Cluster{InstanceTypes: trial.types} - args, err := s.cmd.sbatchArgs(container) - c.Check(err, Equals, trial.err) + args, err := s.disp.sbatchArgs(container) + c.Check(err == nil, Equals, trial.err == nil) if trial.err == nil { - c.Check(args, DeepEquals, append([]string{"--job-name=123", "--mem=239", "--cpus-per-task=2", "--tmp=0", "--nice=9990"}, trial.sbatchArgs...)) + c.Check(args, DeepEquals, append([]string{"--job-name=123", "--nice=10000"}, trial.sbatchArgs...)) + } else { + c.Check(len(err.(dispatchcloud.ConstraintsNotSatisfiableError).AvailableTypes), Equals, len(trial.types)) } } } @@ -419,9 +415,10 @@ func (s *StubbedSuite) TestSbatchPartition(c *C) { Priority: 1, } - args, err := s.cmd.sbatchArgs(container) + args, err := s.disp.sbatchArgs(container) c.Check(args, DeepEquals, []string{ - "--job-name=123", "--mem=239", "--cpus-per-task=1", "--tmp=0", "--nice=9990", + "--job-name=123", "--nice=10000", + "--mem=239", "--cpus-per-task=1", "--tmp=0", "--partition=blurb,b2", }) c.Check(err, IsNil)