import (
"bytes"
"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"
- "math"
"net/http"
"net/http/httptest"
"os"
}
func (s *TestSuite) TestIntegrationNormal(c *C) {
- s.integrationTest(c, false)
+ 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 (s *TestSuite) TestIntegrationCancel(c *C) {
+
+ // 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")
+ }
+
+ 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 (s *TestSuite) TestIntegrationMissingFromSqueue(c *C) {
- s.integrationTest(c, true)
+ 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 (s *TestSuite) integrationTest(c *C, missingFromSqueue bool) {
+func (s *TestSuite) integrationTest(c *C,
+ newSqueueCmd func() *exec.Cmd,
+ sbatchCmdComps []string,
+ runContainer func(*dispatch.Dispatcher, arvados.Container)) arvados.Container {
arvadostest.ResetEnv()
arv, err := arvadosclient.MakeArvadosClient()
var sbatchCmdLine []string
// Override sbatchCmd
- defer func(orig func(dispatch.Container) *exec.Cmd) {
+ defer func(orig func(arvados.Container) *exec.Cmd) {
sbatchCmd = orig
}(sbatchCmd)
- sbatchCmd = func(container dispatch.Container) *exec.Cmd {
+ sbatchCmd = func(container arvados.Container) *exec.Cmd {
sbatchCmdLine = sbatchFunc(container).Args
return exec.Command("sh")
}
defer func(orig func() *exec.Cmd) {
squeueCmd = orig
}(squeueCmd)
- squeueCmd = func() *exec.Cmd {
- if missingFromSqueue {
- return exec.Command("echo")
- } else {
- return exec.Command("echo", "zzzzz-dz642-queuedcontainer")
- }
- }
+ squeueCmd = newSqueueCmd
// There should be no queued containers now
params := arvadosclient.Dict{
- "filters": [][]string{[]string{"state", "=", "Queued"}},
+ "filters": [][]string{{"state", "=", "Queued"}},
}
- var containers dispatch.ContainerList
+ var containers arvados.ContainerList
err = arv.List("containers", params, &containers)
c.Check(err, IsNil)
c.Check(len(containers.Items), Equals, 1)
- echo := "echo"
- crunchRunCommand = &echo
+ theConfig.CrunchRunCommand = []string{"echo"}
doneProcessing := make(chan struct{})
dispatcher := dispatch.Dispatcher{
Arv: arv,
PollInterval: time.Duration(1) * time.Second,
RunContainer: func(dispatcher *dispatch.Dispatcher,
- container dispatch.Container,
- status chan dispatch.Container) {
- go func() {
- dispatcher.UpdateState(container.UUID, dispatch.Running)
- time.Sleep(3 * time.Second)
- dispatcher.UpdateState(container.UUID, dispatch.Complete)
- }()
+ container arvados.Container,
+ status chan arvados.Container) {
+ go runContainer(dispatcher, container)
run(dispatcher, container, status)
doneProcessing <- struct{}{}
},
DoneProcessing: doneProcessing}
- squeueUpdater.SqueueDone = make(chan struct{})
- go squeueUpdater.SyncSqueue(time.Duration(500) * time.Millisecond)
+ squeueUpdater.StartMonitor(time.Duration(500) * time.Millisecond)
err = dispatcher.RunDispatcher()
c.Assert(err, IsNil)
- squeueUpdater.SqueueDone <- struct{}{}
- close(squeueUpdater.SqueueDone)
-
- item := containers.Items[0]
- sbatchCmdComps := []string{"sbatch", "--share", "--parsable",
- fmt.Sprintf("--job-name=%s", item.UUID),
- fmt.Sprintf("--mem-per-cpu=%d", int(math.Ceil(float64(item.RuntimeConstraints["ram"])/float64(item.RuntimeConstraints["vcpus"]*1048576)))),
- fmt.Sprintf("--cpus-per-task=%d", int(item.RuntimeConstraints["vcpus"])),
- fmt.Sprintf("--priority=%d", item.Priority)}
-
- if missingFromSqueue {
- // not in squeue when run() started, so it will have called sbatch
- c.Check(sbatchCmdLine, DeepEquals, sbatchCmdComps)
- } else {
- // already in squeue when run() started, will have just monitored it instead
- c.Check(sbatchCmdLine, DeepEquals, []string(nil))
- }
+ squeueUpdater.Done()
+
+ c.Check(sbatchCmdLine, DeepEquals, sbatchCmdComps)
// There should be no queued containers now
err = arv.List("containers", params, &containers)
c.Check(len(containers.Items), Equals, 0)
// Previously "Queued" container should now be in "Complete" state
- var container dispatch.Container
+ var container arvados.Container
err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container)
c.Check(err, IsNil)
- if missingFromSqueue {
- c.Check(container.State, Equals, "Cancelled")
- } else {
- c.Check(container.State, Equals, "Complete")
- }
+ return container
}
-func (s *MockArvadosServerSuite) Test_APIErrorGettingContainers(c *C) {
+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 + `"}`}
apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
log.SetOutput(io.MultiWriter(buf, os.Stderr))
defer log.SetOutput(os.Stderr)
- crunchRunCommand = &crunchCmd
+ theConfig.CrunchRunCommand = []string{crunchCmd}
doneProcessing := make(chan struct{})
dispatcher := dispatch.Dispatcher{
Arv: arv,
PollInterval: time.Duration(1) * time.Second,
RunContainer: func(dispatcher *dispatch.Dispatcher,
- container dispatch.Container,
- status chan dispatch.Container) {
+ container arvados.Container,
+ status chan arvados.Container) {
go func() {
time.Sleep(1 * time.Second)
dispatcher.UpdateState(container.UUID, dispatch.Running)
c.Check(buf.String(), Matches, `(?ms).*`+expected+`.*`)
}
+
+func (s *MockArvadosServerSuite) TestNoSuchConfigFile(c *C) {
+ var config Config
+ err := readConfig(&config, "/nosuchdir89j7879/8hjwr7ojgyy7")
+ c.Assert(err, NotNil)
+}
+
+func (s *MockArvadosServerSuite) TestBadSbatchArgsConfig(c *C) {
+ var config Config
+
+ tmpfile, err := ioutil.TempFile(os.TempDir(), "config")
+ c.Check(err, IsNil)
+ defer os.Remove(tmpfile.Name())
+
+ _, err = tmpfile.Write([]byte(`{"SbatchArguments": "oops this is not a string array"}`))
+ c.Check(err, IsNil)
+
+ err = readConfig(&config, tmpfile.Name())
+ c.Assert(err, NotNil)
+}
+
+func (s *MockArvadosServerSuite) TestNoSuchArgInConfigIgnored(c *C) {
+ var config Config
+
+ tmpfile, err := ioutil.TempFile(os.TempDir(), "config")
+ c.Check(err, IsNil)
+ defer os.Remove(tmpfile.Name())
+
+ _, err = tmpfile.Write([]byte(`{"NoSuchArg": "Nobody loves me, not one tiny hunk."}`))
+ c.Check(err, IsNil)
+
+ err = readConfig(&config, tmpfile.Name())
+ c.Assert(err, IsNil)
+ c.Check(0, Equals, len(config.SbatchArguments))
+}
+
+func (s *MockArvadosServerSuite) TestReadConfig(c *C) {
+ var config Config
+
+ tmpfile, err := ioutil.TempFile(os.TempDir(), "config")
+ c.Check(err, IsNil)
+ defer os.Remove(tmpfile.Name())
+
+ args := []string{"--arg1=v1", "--arg2", "--arg3=v3"}
+ argsS := `{"SbatchArguments": ["--arg1=v1", "--arg2", "--arg3=v3"]}`
+ _, err = tmpfile.Write([]byte(argsS))
+ c.Check(err, IsNil)
+
+ err = readConfig(&config, 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)
+}
+
+func (s *MockArvadosServerSuite) TestSbatchFuncWithEmptyConfigArgs(c *C) {
+ testSbatchFuncWithArgs(c, []string{})
+}
+
+func (s *MockArvadosServerSuite) TestSbatchFuncWithConfigArgs(c *C) {
+ testSbatchFuncWithArgs(c, []string{"--arg1=v1", "--arg2"})
+}
+
+func testSbatchFuncWithArgs(c *C, args []string) {
+ theConfig.SbatchArguments = append(theConfig.SbatchArguments, args...)
+
+ container := arvados.Container{UUID: "123", RuntimeConstraints: arvados.RuntimeConstraints{RAM: 1000000, 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=1", "--cpus-per-task=2")
+
+ c.Check(sbatchCmd.Args, DeepEquals, expected)
+}
projects / arvados.git / blobdiff