package main
import (
- "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
- "git.curoverse.com/arvados.git/sdk/go/arvadostest"
-
- "io/ioutil"
+ "bytes"
+ "context"
+ "io"
"log"
"net/http"
"net/http/httptest"
"os"
+ "os/exec"
"strings"
"testing"
"time"
+ "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"
)
func (s *TestSuite) SetUpSuite(c *C) {
initialArgs = os.Args
arvadostest.StartAPI()
+ runningCmds = make(map[string]*exec.Cmd)
}
func (s *TestSuite) TearDownSuite(c *C) {
func (s *TestSuite) SetUpTest(c *C) {
args := []string{"crunch-dispatch-local"}
os.Args = args
-
- var err error
- arv, err = arvadosclient.MakeArvadosClient()
- if err != nil {
- c.Fatalf("Error making arvados client: %s", err)
- }
}
func (s *TestSuite) TearDownTest(c *C) {
arvadostest.ResetEnv()
}
-func (s *TestSuite) Test_doMain(c *C) {
- args := []string{"-poll-interval", "1", "-container-priority-poll-interval", "1", "-crunch-run-command", "echo"}
- os.Args = append(os.Args, args...)
+func (s *TestSuite) TestIntegration(c *C) {
+ arv, err := arvadosclient.MakeArvadosClient()
+ c.Assert(err, IsNil)
+
+ echo := "echo"
+ crunchRunCommand = &echo
+
+ ctx, cancel := context.WithCancel(context.Background())
+ dispatcher := dispatch.Dispatcher{
+ Arv: arv,
+ PollPeriod: time.Second,
+ RunContainer: func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
+ run(d, c, s)
+ cancel()
+ },
+ }
- go func() {
- time.Sleep(2 * time.Second)
- doneProcessing <- true
- }()
+ startCmd = func(container arvados.Container, cmd *exec.Cmd) error {
+ dispatcher.UpdateState(container.UUID, "Running")
+ dispatcher.UpdateState(container.UUID, "Complete")
+ return cmd.Start()
+ }
- err := doMain()
- c.Check(err, IsNil)
+ err = dispatcher.Run(ctx)
+ c.Assert(err, Equals, context.Canceled)
+
+ // Wait for all running crunch jobs to complete / terminate
+ waitGroup.Wait()
// There should be no queued containers now
params := arvadosclient.Dict{
- "filters": [][]string{[]string{"state", "=", "Queued"}},
+ "filters": [][]string{{"state", "=", "Queued"}},
}
- var containers ContainerList
+ var containers arvados.ContainerList
err = arv.List("containers", params, &containers)
c.Check(err, IsNil)
- c.Assert(containers.ItemsAvailable, Equals, 0)
+ c.Assert(len(containers.Items), Equals, 0)
// Previously "Queued" container should now be in "Complete" state
- var container Container
+ var container arvados.Container
err = arv.Get("containers", "zzzzz-dz642-queuedcontainer", nil, &container)
c.Check(err, IsNil)
- c.Check(container.State, Equals, "Complete")
+ c.Check(string(container.State), Equals, "Complete")
}
func (s *MockArvadosServerSuite) Test_APIErrorGettingContainers(c *C) {
apiStubResponses := make(map[string]arvadostest.StubResponse)
apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
- testWithServerStub(c, apiStubResponses, "echo", "Error getting list of queued containers")
+ testWithServerStub(c, apiStubResponses, "echo", "Error getting list of containers")
}
func (s *MockArvadosServerSuite) Test_APIErrorUpdatingContainerState(c *C) {
apiStubResponses := make(map[string]arvadostest.StubResponse)
apiStubResponses["/arvados/v1/containers"] =
- arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx1"}]}`)}
+ arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx1","State":"Queued","Priority":1}]}`)}
apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx1"] =
arvadostest.StubResponse{500, string(`{}`)}
- testWithServerStub(c, apiStubResponses, "echo", "Error updating container state")
+ testWithServerStub(c, apiStubResponses, "echo", "error locking container zzzzz-dz642-xxxxxxxxxxxxxx1")
}
func (s *MockArvadosServerSuite) Test_ContainerStillInRunningAfterRun(c *C) {
apiStubResponses := make(map[string]arvadostest.StubResponse)
apiStubResponses["/arvados/v1/containers"] =
- arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2"}]}`)}
+ arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2","State":"Queued","Priority":1}]}`)}
+ apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2/lock"] =
+ arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Locked", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx2"] =
- arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1}`)}
+ arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
testWithServerStub(c, apiStubResponses, "echo",
- "After crunch-run process termination, the state is still 'Running' for zzzzz-dz642-xxxxxxxxxxxxxx2")
+ `After echo process termination, container state for Running is "zzzzz-dz642-xxxxxxxxxxxxxx2". Updating it to "Cancelled"`)
}
func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {
apiStubResponses := make(map[string]arvadostest.StubResponse)
apiStubResponses["/arvados/v1/containers"] =
- arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3"}]}`)}
- apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx3"] =
- arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3", "state":"Running", "priority":1}`)}
+ arvadostest.StubResponse{200, string(`{"items_available":1, "items":[{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3","State":"Queued","Priority":1}]}`)}
+
+ apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx3/lock"] =
+ arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3", "state":"Locked", "priority":1}`)}
- testWithServerStub(c, apiStubResponses, "nosuchcommand", "Error running container for zzzzz-dz642-xxxxxxxxxxxxxx3")
+ testWithServerStub(c, apiStubResponses, "nosuchcommand", "Error starting nosuchcommand for zzzzz-dz642-xxxxxxxxxxxxxx3")
}
func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) {
+ apiStubResponses["/arvados/v1/api_client_authorizations/current"] =
+ arvadostest.StubResponse{200, string(`{"uuid": "` + arvadostest.Dispatch1AuthUUID + `", "api_token": "xyz"}`)}
+
apiStub := arvadostest.ServerStub{apiStubResponses}
api := httptest.NewServer(&apiStub)
defer api.Close()
- arv = arvadosclient.ArvadosClient{
+ arv := &arvadosclient.ArvadosClient{
Scheme: "http",
ApiServer: api.URL[7:],
ApiToken: "abc123",
Retries: 0,
}
- tempfile, err := ioutil.TempFile(os.TempDir(), "temp-log-file")
- c.Check(err, IsNil)
- defer os.Remove(tempfile.Name())
- log.SetOutput(tempfile)
+ buf := bytes.NewBuffer(nil)
+ log.SetOutput(io.MultiWriter(buf, os.Stderr))
+ defer log.SetOutput(os.Stderr)
+
+ *crunchRunCommand = crunchCmd
+
+ ctx, cancel := context.WithCancel(context.Background())
+ dispatcher := dispatch.Dispatcher{
+ Arv: arv,
+ PollPeriod: time.Duration(1) * time.Second,
+ RunContainer: func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
+ run(d, c, s)
+ cancel()
+ },
+ }
+
+ startCmd = func(container arvados.Container, cmd *exec.Cmd) error {
+ dispatcher.UpdateState(container.UUID, "Running")
+ dispatcher.UpdateState(container.UUID, "Complete")
+ return cmd.Start()
+ }
go func() {
- time.Sleep(1 * time.Second)
- doneProcessing <- true
+ for i := 0; i < 80 && !strings.Contains(buf.String(), expected); i++ {
+ time.Sleep(100 * time.Millisecond)
+ }
+ cancel()
}()
- runQueuedContainers(1, 1, crunchCmd)
+ err := dispatcher.Run(ctx)
+ c.Assert(err, Equals, context.Canceled)
- // Give some time for run goroutine to complete
- time.Sleep(5 * time.Second)
+ // Wait for all running crunch jobs to complete / terminate
+ waitGroup.Wait()
- buf, _ := ioutil.ReadFile(tempfile.Name())
- c.Check(strings.Contains(string(buf), expected), Equals, true)
+ c.Check(buf.String(), Matches, `(?ms).*`+expected+`.*`)
}