Can set log file for listeners

Author: George Malamidis

client.go

@@ -5,7 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/nutrun/lentil"
-	"log"
+	"os"
 	"strings"
 )
 
@@ -28,7 +28,7 @@ func NewClient(verbose bool) (*Client, error) {
 func (this *Client) put(msg *Message) error {
 	message, e := json.Marshal(msg)
 	if this.verbose {
-		log.Printf("QUEUEING UP: %s\n", message)
+		fmt.Fprintf(os.Stderr, "QUEUEING UP: %s\n", message)
 	}
 	if e != nil {
 		return e
@@ -98,7 +98,7 @@ func (this *Client) drain(tubes string) error {
 		}
 	}
 	drainedJobs = append(drainedJobs, []byte("\n]")...)
-	log.Printf("%s", string(drainedJobs))
+	fmt.Fprintf(os.Stderr, "%s", string(drainedJobs))
 	return nil
 }
 
@@ -108,7 +108,7 @@ func (this *Client) pause(tubes string, delay int) error {
 		if e != nil {
 			return e
 		}
-		log.Printf("Paused %s for %d seconds", tubes, delay)
+		fmt.Fprintf(os.Stderr, "Paused %s for %d seconds", tubes, delay)
 	}
 	return nil
 }

integration_test.py

@@ -7,6 +7,7 @@
 from select import select
 
 HERE = os.path.dirname(os.path.abspath(__file__))
+debug = False
 
 class TestGlowIntegration(unittest.TestCase):
 
@@ -131,7 +132,7 @@ def test_append_to_output_file_if_exists(self):
             self.assertEqual('job1\njob2\n', outfile.read())
         self.listener.interrupt()    
 
-debug = False
+
 
 class Listener:
     def __init__(self):
@@ -171,7 +172,8 @@ def _wait_for_job_update(self, job_desc, status, seconds, max_num_non_matching_e
             fds, _, _ = select([self.process.stderr], [], [], seconds)
             if fds != [self.process.stderr]:
                 raise Exception('timed out waiting for {0} {1}'.format(status, job_desc))
-            line = self.process.stderr.readline()
+            line = self.process.stderr.readline().split(' ')[2:] # get rid of log date/time
+            line = " ".join(line)
             if debug: print line
             if line.startswith(status):
                 job = cjson.decode(line[len(status):])

listener.go

@@ -18,14 +18,17 @@ type Listener struct {
 	sig      os.Signal
 }
 
-func NewListener(verbose, inclusive bool, filter []string) (*Listener, error) {
+func NewListener(verbose, inclusive bool, filter []string, logpath string) (*Listener, error) {
 	this := new(Listener)
+	logfile, err := os.OpenFile(logpath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0666)
+	if err != nil {
+		return nil, err
+	}
+	this.logger = log.New(logfile, "", log.LstdFlags)
 	q, err := lentil.Dial(Config.QueueAddr)
-
 	if err != nil {
 		return nil, err
 	}
-
 	this.q = q
 	this.verbose = verbose
 	this.jobqueue = NewJobQueue(this.q, inclusive, filter)
@@ -42,18 +45,18 @@ listenerloop:
 		}
 		e := Config.LoadDeps()
 		if e != nil {
-			log.Fatalf("Error loading dependency config: %s\n", e)
+			this.logger.Fatalf("Error loading dependency config: %s\n", e)
 		}
 		job, e := this.jobqueue.Next()
 		if e != nil {
 			if strings.Contains(e.Error(), "TIMED_OUT") {
 				time.Sleep(time.Second)
 				goto listenerloop
 			}
-			log.Fatal(e)
+			this.logger.Fatal(e)
 		}
 		if this.verbose {
-			log.Printf("RUNNING: %s", job.Body)
+			this.logger.Printf("RUNNING: %s", job.Body)
 		}
 		msg := new(Message)
 		e = json.Unmarshal([]byte(job.Body), &msg)
@@ -63,10 +66,10 @@ listenerloop:
 		e = this.execute(msg)
 		if e == nil {
 			if this.verbose {
-				log.Printf("COMPLETE: %s", job.Body)
+				this.logger.Printf("COMPLETE: %s", job.Body)
 			}
 		} else {
-			log.Printf("FAILED: %s", job.Body)
+			this.logger.Printf("FAILED: %s", job.Body)
 		}
 		e = this.jobqueue.Delete(job.Id)
 		if e != nil {
@@ -81,12 +84,12 @@ func (this *Listener) trap() {
 	signal.Notify(receivedSignal, syscall.SIGTERM, syscall.SIGINT)
 	this.sig = <-receivedSignal
 	if this.sig.String() == syscall.SIGTERM.String() {
-		log.Printf("Got signal %d. Killing current job.", this.sig)
+		this.logger.Printf("Got signal %d. Killing current job.", this.sig)
 		if this.proc != nil {
 			this.proc.Kill()
 		}
 		os.Exit(1)
 	}
 	go this.trap()
-	log.Printf("Got signal %d. Waiting for current job to complete.\n", this.sig)
+	this.logger.Printf("Got signal %d. Waiting for current job to complete.\n", this.sig)
 }

listener_test.go

@@ -1,10 +1,8 @@
 package main
 
 import (
-	"bytes"
 	"encoding/json"
 	"io/ioutil"
-	"log"
 	"os"
 	"strings"
 	"testing"
@@ -36,11 +34,10 @@ func TestOutput(t *testing.T) {
 }
 
 func TestPutErrorOnBeanstalk(t *testing.T) {
-	listener, err := NewListener(false, false, []string{})
+	listener, err := NewListener(false, false, []string{}, "/dev/null")
 	if err != nil {
 		t.Fatal(err)
 	}
-	log.SetOutput(bytes.NewBufferString(""))
 	msg, e := createTestMessage("lsdonmybrain", "test.out", ".")
 	if e != nil {
 		t.Fatal(e)

main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bufio"
 	"flag"
+	"fmt"
 	"log"
 	"os"
 	"strings"
@@ -27,11 +28,11 @@ var pausedelay *int = flag.Int("pause-delay", 0, "How many seconds to pause tube
 var mailfrom *string = flag.String("mail-from", "[email protected]", "Email 'from' field [listen]")
 var smtpserver *string = flag.String("smtp-server", "", "Server to use for sending emails [listen]")
 var deps *string = flag.String("deps", "", "Path to tube dependency config file [listen]")
+var logpath *string = flag.String("log", "/dev/stderr", "Path to log file [listen]")
 
 var Config *Configuration
 
 func main() {
-	log.SetFlags(0)
 	flag.Parse()
 	Config = NewConfig(*deps, *smtpserver, *mailfrom)
 	if *listener {
@@ -40,9 +41,10 @@ func main() {
 		if *exclude != "" {
 			filter = strings.Split(*exclude, ",")
 		}
-		l, e := NewListener(*verbose, include, filter)
+		l, e := NewListener(*verbose, include, filter, *logpath)
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 		l.run()
 		return
@@ -56,38 +58,50 @@ func main() {
 		// hack: local doesn't need tube, defaulting it to respect the Message API
 		msg, e := NewMessage(executable, arguments, *mailto, *workdir, *stdout, *stderr, "localignore", 0, 0)
 		if e != nil {
-			log.Fatal(e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
-		runner, e := NewRunner(*verbose)
+		logfile, e := os.OpenFile(*logpath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0666)
 		if e != nil {
-			log.Fatal(e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
+		}
+		runner, e := NewRunner(*verbose, log.New(logfile, "", log.LstdFlags))
+		if e != nil {
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 		e = runner.execute(msg)
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 		return
 	}
 
 	c, e := NewClient(*verbose)
 	if e != nil {
-		log.Fatalf("ERROR: %s", e)
+		fmt.Fprintln(os.Stderr, e)
+		os.Exit(1)
 	}
 
 	if *drain != "" {
 		e = c.drain(*drain)
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 	} else if *pause != "" {
 		if *pausedelay == 0 {
-			log.Fatal("Usage: glow -pause=<tube1,tube2,...> -pause-delay=<seconds>")
+			fmt.Fprintln(os.Stderr, "Usage: glow -pause=<tube1,tube2,...> -pause-delay=<seconds>")
+			os.Exit(1)
 		}
 		e = c.pause(*pause, *pausedelay)
 	} else if *stats {
 		e = c.stats()
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 	} else if len(flag.Args()) == 0 { // Queue up many jobs from STDIN
 		in := bufio.NewReaderSize(os.Stdin, 1024*1024)
@@ -98,23 +112,27 @@ func main() {
 				if e.Error() == "EOF" {
 					break
 				}
-				log.Fatalf("ERROR: %s", e)
+				fmt.Fprintln(os.Stderr, e)
+				os.Exit(1)
 			}
 			input = append(input, line...)
 		}
 		e = c.putMany(input)
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 	} else { // Queue up one job
 		executable, arguments := parseCommand()
 		msg, e := NewMessage(executable, arguments, *mailto, *workdir, *stdout, *stderr, *tube, *priority, *delay)
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 		e = c.put(msg)
 		if e != nil {
-			log.Fatalf("ERROR: %s", e)
+			fmt.Fprintln(os.Stderr, e)
+			os.Exit(1)
 		}
 	}
 }

runner.go

@@ -16,10 +16,12 @@ type Runner struct {
 	q       *lentil.Beanstalkd
 	proc    *os.Process
 	verbose bool
+	logger  *log.Logger
 }
 
-func NewRunner(verbose bool) (*Runner, error) {
+func NewRunner(verbose bool, logger *log.Logger) (*Runner, error) {
 	this := new(Runner)
+	this.logger = logger
 	q, err := lentil.Dial(Config.QueueAddr)
 
 	if err != nil {
@@ -60,9 +62,9 @@ func (this *Runner) execute(msg *Message) error {
 	cmd.Stdout = stdoutF
 	cmd.Stderr = stderrF
 	if this.verbose {
-		log.Printf("INFO: Running command '%s %s'\n", msg.Executable, strings.Join(msg.Arguments, " "))
-		log.Printf("INFO: STDOUT to %s\n", msg.Stdout)
-		log.Printf("INFO: STDERR to %s\n", msg.Stderr)
+		this.logger.Printf("INFO: Running command '%s %s'\n", msg.Executable, strings.Join(msg.Arguments, " "))
+		this.logger.Printf("INFO: STDOUT to %s\n", msg.Stdout)
+		this.logger.Printf("INFO: STDERR to %s\n", msg.Stderr)
 	}
 	e = cmd.Start()
 	if e != nil {
@@ -80,25 +82,25 @@ func (this *Runner) execute(msg *Message) error {
 
 // Log and email errors
 func (this *Runner) catch(msg *Message, e error) {
-	log.Printf("ERROR: %s\n", e)
+	this.logger.Printf("ERROR: %s\n", e)
 	this.mail(msg, e)
 	this.publishError(msg, e)
 }
 
 func (this *Runner) publishError(msg *Message, e error) {
 	err := this.q.Use(Config.errorQueue)
 	if err != nil {
-		log.Printf("ERROR: %s\n", err)
+		this.logger.Printf("ERROR: %s\n", err)
 		return
 	}
 	payload, err := json.Marshal(NewErrMessage(msg, e))
 	if err != nil {
-		log.Printf("ERROR: %s\n", err)
+		this.logger.Printf("ERROR: %s\n", err)
 		return
 	}
 	_, err = this.q.Put(0, 0, 60*60, payload)
 	if err != nil {
-		log.Printf("ERROR: %s\n", err)
+		this.logger.Printf("ERROR: %s\n", err)
 	}
 }
 
@@ -115,6 +117,6 @@ func (this *Runner) mail(msg *Message, e error) {
 	mail := fmt.Sprintf("%s%s", subject, fmt.Sprintf("Ran on [%s]\n%s\n%s\n%s", hostname, subject, e, msg.readOut()))
 	e = smtp.SendMail(Config.SmtpServer, nil, Config.MailFrom, to, []byte(mail))
 	if e != nil {
-		log.Printf("ERROR: %s\n", e)
+		this.logger.Printf("ERROR: %s\n", e)
 	}
 }

runner_test.go

@@ -10,7 +10,15 @@ import (
 )
 
 func TestRunnerOutput(t *testing.T) {
-	runner := new(Runner)
+	devnull, e := os.OpenFile(os.DevNull, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0666)
+	if e != nil {
+		t.Fatal(e)
+	}
+	l := log.New(devnull, "", log.LstdFlags)
+	runner, e := NewRunner(false, l)
+	if e != nil {
+		t.Fatal(e)
+	}
 	msg, e := createTestMessage("echo you suck", "test.out", ".")
 	if e != nil {
 		t.Fatal(e)
@@ -30,7 +38,12 @@ func TestRunnerOutput(t *testing.T) {
 }
 
 func TestRunnerShouldPutErrorOnBeanstalk(t *testing.T) {
-	runner, err := NewRunner(false)
+	devnull, err := os.OpenFile(os.DevNull, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0666)
+	if err != nil {
+		t.Fatal(err)
+	}
+	l := log.New(devnull, "", log.LstdFlags)
+	runner, err := NewRunner(false, l)
 	if err != nil {
 		t.Fatal(err)
 	}