13959: Use logrus for crunch-dispatch-local logging.

Arvados-DCO-1.1-Signed-off-by: Tom Clegg <tclegg@veritasgenetics.com>

diff --git a/sdk/go/dispatch/dispatch.go b/sdk/go/dispatch/dispatch.go
index 74cefed05794b7c242ebd8033fba1e8a5781547a..152207ea94bb4bb4c4e21de36ca2b440691109a2 100644 (file)
--- a/sdk/go/dispatch/dispatch.go
+++ b/sdk/go/dispatch/dispatch.go
@@ -230,7 +230,7 @@ func (d *Dispatcher) checkListForUpdates(containers []arvados.Container, todo ma
                                }
                                err := d.lock(c.UUID)
                                if err != nil {
                                }
                                err := d.lock(c.UUID)
                                if err != nil {

-                                       d.Logger.Debugf("error locking container %s: %s", c.UUID, err)
+                                       d.Logger.Warnf("error locking container %s: %s", c.UUID, err)

                                        break
                                }
                                c.State = Locked
                                        break
                                }
                                c.State = Locked

diff --git a/services/crunch-dispatch-local/crunch-dispatch-local.go b/services/crunch-dispatch-local/crunch-dispatch-local.go
index 279327ba18811ba8ad6339600cc124460f2fc35c..fc10393626be103c17b01b5b1bfde615ed470bc9 100644 (file)
--- a/services/crunch-dispatch-local/crunch-dispatch-local.go
+++ b/services/crunch-dispatch-local/crunch-dispatch-local.go
@@ -10,7 +10,6 @@ import (
        "context"
        "flag"
        "fmt"
        "context"
        "flag"
        "fmt"

-       "log"

        "os"
        "os/exec"
        "os/signal"
        "os"
        "os/exec"
        "os/signal"


@@ -21,6 +20,7 @@ import (
        "git.curoverse.com/arvados.git/sdk/go/arvados"
        "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
        "git.curoverse.com/arvados.git/sdk/go/dispatch"
        "git.curoverse.com/arvados.git/sdk/go/arvados"
        "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
        "git.curoverse.com/arvados.git/sdk/go/dispatch"

+       "github.com/Sirupsen/logrus"

 )
 
 var version = "dev"
 )
 
 var version = "dev"


@@ -28,7 +28,7 @@ var version = "dev"
 func main() {
        err := doMain()
        if err != nil {
 func main() {
        err := doMain()
        if err != nil {

-               log.Fatalf("%q", err)
+               logrus.Fatalf("%q", err)

        }
 }
 
        }
 }
 


@@ -40,6 +40,14 @@ var (
 )
 
 func doMain() error {
 )
 
 func doMain() error {

+       logger := logrus.StandardLogger()
+       if os.Getenv("DEBUG") != "" {
+               logger.SetLevel(logrus.DebugLevel)
+       }
+       logger.Formatter = &logrus.JSONFormatter{
+               TimestampFormat: "2006-01-02T15:04:05.000000000Z07:00",
+       }
+

        flags := flag.NewFlagSet("crunch-dispatch-local", flag.ExitOnError)
 
        pollInterval := flags.Int(
        flags := flag.NewFlagSet("crunch-dispatch-local", flag.ExitOnError)
 
        pollInterval := flags.Int(


@@ -66,18 +74,19 @@ func doMain() error {
                return nil
        }
 
                return nil
        }
 

-       log.Printf("crunch-dispatch-local %s started", version)
+       logger.Printf("crunch-dispatch-local %s started", version)

 
        runningCmds = make(map[string]*exec.Cmd)
 
        arv, err := arvadosclient.MakeArvadosClient()
        if err != nil {
 
        runningCmds = make(map[string]*exec.Cmd)
 
        arv, err := arvadosclient.MakeArvadosClient()
        if err != nil {

-               log.Printf("Error making Arvados client: %v", err)
+               logger.Errorf("error making Arvados client: %v", err)

                return err
        }
        arv.Retries = 25
 
        dispatcher := dispatch.Dispatcher{
                return err
        }
        arv.Retries = 25
 
        dispatcher := dispatch.Dispatcher{

+               Logger:       logger,

                Arv:          arv,
                RunContainer: run,
                PollPeriod:   time.Duration(*pollInterval) * time.Second,
                Arv:          arv,
                RunContainer: run,
                PollPeriod:   time.Duration(*pollInterval) * time.Second,


@@ -92,7 +101,7 @@ func doMain() error {
        c := make(chan os.Signal, 1)
        signal.Notify(c, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)
        sig := <-c
        c := make(chan os.Signal, 1)
        signal.Notify(c, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)
        sig := <-c

-       log.Printf("Received %s, shutting down", sig)
+       logger.Printf("Received %s, shutting down", sig)

        signal.Stop(c)
 
        cancel()
        signal.Stop(c)
 
        cancel()


@@ -138,7 +147,7 @@ func run(dispatcher *dispatch.Dispatcher,
                cmd.Stderr = os.Stderr
                cmd.Stdout = os.Stderr
 
                cmd.Stderr = os.Stderr
                cmd.Stdout = os.Stderr
 

-               log.Printf("Starting container %v", uuid)
+               dispatcher.Logger.Printf("starting container %v", uuid)

 
                // Add this crunch job to the list of runningCmds only if we
                // succeed in starting crunch-run.
 
                // Add this crunch job to the list of runningCmds only if we
                // succeed in starting crunch-run.


@@ -146,7 +155,7 @@ func run(dispatcher *dispatch.Dispatcher,
                runningCmdsMutex.Lock()
                if err := startCmd(container, cmd); err != nil {
                        runningCmdsMutex.Unlock()
                runningCmdsMutex.Lock()
                if err := startCmd(container, cmd); err != nil {
                        runningCmdsMutex.Unlock()

-                       log.Printf("Error starting %v for %v: %q", *crunchRunCommand, uuid, err)
+                       dispatcher.Logger.Warnf("error starting %q for %s: %s", *crunchRunCommand, uuid, err)

                        dispatcher.UpdateState(uuid, dispatch.Cancelled)
                } else {
                        runningCmds[uuid] = cmd
                        dispatcher.UpdateState(uuid, dispatch.Cancelled)
                } else {
                        runningCmds[uuid] = cmd


@@ -157,9 +166,9 @@ func run(dispatcher *dispatch.Dispatcher,
 
                        go func() {
                                if _, err := cmd.Process.Wait(); err != nil {
 
                        go func() {
                                if _, err := cmd.Process.Wait(); err != nil {

-                                       log.Printf("Error while waiting for crunch job to finish for %v: %q", uuid, err)
+                                       dispatcher.Logger.Warnf("error while waiting for crunch job to finish for %v: %q", uuid, err)

                                }
                                }

-                               log.Printf("sending done")
+                               dispatcher.Logger.Debugf("sending done")

                                done <- struct{}{}
                        }()
 
                                done <- struct{}{}
                        }()
 


@@ -171,14 +180,14 @@ func run(dispatcher *dispatch.Dispatcher,
                                case c := <-status:
                                        // Interrupt the child process if priority changes to 0
                                        if (c.State == dispatch.Locked || c.State == dispatch.Running) && c.Priority == 0 {
                                case c := <-status:
                                        // Interrupt the child process if priority changes to 0
                                        if (c.State == dispatch.Locked || c.State == dispatch.Running) && c.Priority == 0 {

-                                               log.Printf("Sending SIGINT to pid %d to cancel container %v", cmd.Process.Pid, uuid)
+                                               dispatcher.Logger.Printf("sending SIGINT to pid %d to cancel container %v", cmd.Process.Pid, uuid)

                                                cmd.Process.Signal(os.Interrupt)
                                        }
                                }
                        }
                        close(done)
 
                                                cmd.Process.Signal(os.Interrupt)
                                        }
                                }
                        }
                        close(done)
 

-                       log.Printf("Finished container run for %v", uuid)
+                       dispatcher.Logger.Printf("finished container run for %v", uuid)

 
                        // Remove the crunch job from runningCmds
                        runningCmdsMutex.Lock()
 
                        // Remove the crunch job from runningCmds
                        runningCmdsMutex.Lock()


@@ -191,11 +200,11 @@ func run(dispatcher *dispatch.Dispatcher,
        // If the container is not finalized, then change it to "Cancelled".
        err := dispatcher.Arv.Get("containers", uuid, nil, &container)
        if err != nil {
        // If the container is not finalized, then change it to "Cancelled".
        err := dispatcher.Arv.Get("containers", uuid, nil, &container)
        if err != nil {

-               log.Printf("Error getting final container state: %v", err)
+               dispatcher.Logger.Warnf("error getting final container state: %v", err)

        }
        if container.State == dispatch.Locked || container.State == dispatch.Running {
        }
        if container.State == dispatch.Locked || container.State == dispatch.Running {

-               log.Printf("After %s process termination, container state for %v is %q.  Updating it to %q",
-                       *crunchRunCommand, container.State, uuid, dispatch.Cancelled)
+               dispatcher.Logger.Warnf("after %q process termination, container state for %v is %q; updating it to %q",
+                       *crunchRunCommand, uuid, container.State, dispatch.Cancelled)

                dispatcher.UpdateState(uuid, dispatch.Cancelled)
        }
 
                dispatcher.UpdateState(uuid, dispatch.Cancelled)
        }
 


@@ -203,5 +212,5 @@ func run(dispatcher *dispatch.Dispatcher,
        for range status {
        }
 
        for range status {
        }
 

-       log.Printf("Finalized container %v", uuid)
+       dispatcher.Logger.Printf("finalized container %v", uuid)

 }
 }

diff --git a/services/crunch-dispatch-local/crunch-dispatch-local_test.go b/services/crunch-dispatch-local/crunch-dispatch-local_test.go
index 1a2787c25c625d3af04ab51655879ab13c9cbf82..a1a6d5641e01b70ae92b31183218ad8ea28608d7 100644 (file)
--- a/services/crunch-dispatch-local/crunch-dispatch-local_test.go
+++ b/services/crunch-dispatch-local/crunch-dispatch-local_test.go
@@ -8,7 +8,6 @@ import (
        "bytes"
        "context"
        "io"
        "bytes"
        "context"
        "io"

-       "log"

        "net/http"
        "net/http/httptest"
        "os"
        "net/http"
        "net/http/httptest"
        "os"


@@ -21,6 +20,7 @@ import (
        "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/sdk/go/arvadosclient"
        "git.curoverse.com/arvados.git/sdk/go/arvadostest"
        "git.curoverse.com/arvados.git/sdk/go/dispatch"

+       "github.com/Sirupsen/logrus"

        . "gopkg.in/check.v1"
 )
 
        . "gopkg.in/check.v1"
 )
 


@@ -110,7 +110,7 @@ func (s *MockArvadosServerSuite) Test_APIErrorGettingContainers(c *C) {
        apiStubResponses := make(map[string]arvadostest.StubResponse)
        apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
 
        apiStubResponses := make(map[string]arvadostest.StubResponse)
        apiStubResponses["/arvados/v1/containers"] = arvadostest.StubResponse{500, string(`{}`)}
 

-       testWithServerStub(c, apiStubResponses, "echo", "Error getting list of containers")
+       testWithServerStub(c, apiStubResponses, "echo", "error getting list of containers")

 }
 
 func (s *MockArvadosServerSuite) Test_APIErrorUpdatingContainerState(c *C) {
 }
 
 func (s *MockArvadosServerSuite) Test_APIErrorUpdatingContainerState(c *C) {


@@ -133,7 +133,7 @@ func (s *MockArvadosServerSuite) Test_ContainerStillInRunningAfterRun(c *C) {
                arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
 
        testWithServerStub(c, apiStubResponses, "echo",
                arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx2", "state":"Running", "priority":1, "locked_by_uuid": "` + arvadostest.Dispatch1AuthUUID + `"}`)}
 
        testWithServerStub(c, apiStubResponses, "echo",

-               `After echo process termination, container state for Running is "zzzzz-dz642-xxxxxxxxxxxxxx2".  Updating it to "Cancelled"`)
+               `after "echo" process termination, container state for zzzzz-dz642-xxxxxxxxxxxxxx2 is "Running"; updating it to "Cancelled"`)

 }
 
 func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {
 }
 
 func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {


@@ -144,7 +144,7 @@ func (s *MockArvadosServerSuite) Test_ErrorRunningContainer(c *C) {
        apiStubResponses["/arvados/v1/containers/zzzzz-dz642-xxxxxxxxxxxxxx3/lock"] =
                arvadostest.StubResponse{200, string(`{"uuid":"zzzzz-dz642-xxxxxxxxxxxxxx3", "state":"Locked", "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 starting nosuchcommand 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) {
 }
 
 func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubResponse, crunchCmd string, expected string) {


@@ -165,15 +165,15 @@ func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubRespon
        }
 
        buf := bytes.NewBuffer(nil)
        }
 
        buf := bytes.NewBuffer(nil)

-       log.SetOutput(io.MultiWriter(buf, os.Stderr))
-       defer log.SetOutput(os.Stderr)
+       logrus.SetOutput(io.MultiWriter(buf, os.Stderr))
+       defer logrus.SetOutput(os.Stderr)

 
        *crunchRunCommand = crunchCmd
 
        ctx, cancel := context.WithCancel(context.Background())
        dispatcher := dispatch.Dispatcher{
                Arv:        arv,
 
        *crunchRunCommand = crunchCmd
 
        ctx, cancel := context.WithCancel(context.Background())
        dispatcher := dispatch.Dispatcher{
                Arv:        arv,

-               PollPeriod: time.Duration(1) * time.Second,
+               PollPeriod: time.Second / 20,

                RunContainer: func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
                        run(d, c, s)
                        cancel()
                RunContainer: func(d *dispatch.Dispatcher, c arvados.Container, s <-chan arvados.Container) {
                        run(d, c, s)
                        cancel()