From 58044098495d066effa7fd4742b6635d9a10fdfb Mon Sep 17 00:00:00 2001 From: Tom Clegg Date: Fri, 17 Aug 2018 10:19:29 -0400 Subject: [PATCH] 13959: Use logrus for crunch-dispatch-local logging. Arvados-DCO-1.1-Signed-off-by: Tom Clegg --- sdk/go/dispatch/dispatch.go | 2 +- .../crunch-dispatch-local.go | 39 ++++++++++++------- .../crunch-dispatch-local_test.go | 14 +++---- 3 files changed, 32 insertions(+), 23 deletions(-) diff --git a/sdk/go/dispatch/dispatch.go b/sdk/go/dispatch/dispatch.go index 74cefed057..152207ea94 100644 --- 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 { - 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 diff --git a/services/crunch-dispatch-local/crunch-dispatch-local.go b/services/crunch-dispatch-local/crunch-dispatch-local.go index 279327ba18..fc10393626 100644 --- 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" - "log" "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" + "github.com/Sirupsen/logrus" ) var version = "dev" @@ -28,7 +28,7 @@ var version = "dev" func main() { err := doMain() if err != nil { - log.Fatalf("%q", err) + logrus.Fatalf("%q", err) } } @@ -40,6 +40,14 @@ var ( ) 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( @@ -66,18 +74,19 @@ func doMain() error { 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 { - log.Printf("Error making Arvados client: %v", err) + logger.Errorf("error making Arvados client: %v", err) return err } arv.Retries = 25 dispatcher := dispatch.Dispatcher{ + Logger: logger, 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 - log.Printf("Received %s, shutting down", sig) + logger.Printf("Received %s, shutting down", sig) signal.Stop(c) cancel() @@ -138,7 +147,7 @@ func run(dispatcher *dispatch.Dispatcher, 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. @@ -146,7 +155,7 @@ func run(dispatcher *dispatch.Dispatcher, 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 @@ -157,9 +166,9 @@ func run(dispatcher *dispatch.Dispatcher, 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{}{} }() @@ -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 { - 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) - 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() @@ -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 { - 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 { - 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) } @@ -203,5 +212,5 @@ func run(dispatcher *dispatch.Dispatcher, 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 1a2787c25c..a1a6d5641e 100644 --- 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" - "log" "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" + "github.com/Sirupsen/logrus" . "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(`{}`)} - 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) { @@ -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", - `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) { @@ -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}`)} - 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) { @@ -165,15 +165,15 @@ func testWithServerStub(c *C, apiStubResponses map[string]arvadostest.StubRespon } 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, - 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() -- 2.30.2