v0.9.1

Created "params" type to avoid positional arguments.
Personally it feels more explicit and reads better.

CHANGELOG.md

@@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.9.1] - 2020-05-29
+
+### Added
+
+- `GetTaskID`, `GetRetryCount`, and `GetMaxRetry` functions were added to extract task metadata from context.
+
+## [0.9.0] - 2020-05-16
+
+### Changed
+
+- `Logger` interface has changed. Please see the godoc for the new interface.
+
+### Added
+
+- `LogLevel` type is added. Server's log level can be specified through `LogLevel` field in `Config`.
+
+## [0.8.3] - 2020-05-08
+
+### Added
+
+- `Close` method is added to `Client`.
+
 ## [0.8.2] - 2020-05-03
 
 ### Fixed

README.md

@@ -68,9 +68,10 @@ const (
     ImageProcessing = "image:process"
 )
 
-//--------------------------------------------
-// Write function NewXXXTask to create a task.
-//--------------------------------------------
+//----------------------------------------------
+// Write a function NewXXXTask to create a task.
+// A task consists of a type and a payload.
+//----------------------------------------------
 
 func NewEmailDeliveryTask(userID int, tmplID string) *asynq.Task {
     payload := map[string]interface{}{"user_id": userID, "template_id": tmplID}
@@ -82,13 +83,13 @@ func NewImageProcessingTask(src, dst string) *asynq.Task {
     return asynq.NewTask(ImageProcessing, payload)
 }
 
-//-------------------------------------------------------------
-// Write function HandleXXXTask to handle the given task.
-// NOTE: It satisfies the asynq.HandlerFunc interface.
+//---------------------------------------------------------------
+// Write a function HandleXXXTask to handle the input task.
+// Note that it satisfies the asynq.HandlerFunc interface.
 // 
 // Handler doesn't need to be a function. You can define a type 
-// that satisfies asynq.Handler interface. See example below.
-//-------------------------------------------------------------
+// that satisfies asynq.Handler interface. See examples below.
+//---------------------------------------------------------------
 
 func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
     userID, err := t.Payload.GetInt("user_id")
@@ -104,11 +105,11 @@ func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
     return nil
 }
 
+// ImageProcessor implements asynq.Handler interface.
 type ImageProcesser struct {
     // ... fields for struct
 }
 
-// ImageProcessor implements asynq.Handler.
 func (p *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
     src, err := t.Payload.GetString("src")
     if err != nil {
@@ -147,11 +148,12 @@ const redisAddr = "127.0.0.1:6379"
 func main() {
     r := asynq.RedisClientOpt{Addr: redisAddr}
     c := asynq.NewClient(r)
+    defer c.Close()
 
-    // ----------------------------------------------------
+    // ------------------------------------------------------
     // Example 1: Enqueue task to be processed immediately.
     //            Use (*Client).Enqueue method.
-    // ---------------------------------------------------- 
+    // ------------------------------------------------------
 
     t := tasks.NewEmailDeliveryTask(42, "some:template:id")
     err := c.Enqueue(t)
@@ -160,10 +162,10 @@ func main() {
     }
 
 
-    // ----------------------------------------------------------
+    // ------------------------------------------------------------
     // Example 2: Schedule task to be processed in the future.
     //            Use (*Client).EnqueueIn or (*Client).EnqueueAt.
-    // ----------------------------------------------------------
+    // ------------------------------------------------------------
 
     t = tasks.NewEmailDeliveryTask(42, "other:template:id")
     err = c.EnqueueIn(24*time.Hour, t)
@@ -172,10 +174,10 @@ func main() {
     }
 
 
-    // --------------------------------------------------------------------------
+    // ----------------------------------------------------------------------------
     // Example 3: Set options to tune task processing behavior.
     //            Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
-    // --------------------------------------------------------------------------
+    // ----------------------------------------------------------------------------
 
     c.SetDefaultOptions(tasks.ImageProcessing, asynq.MaxRetry(10), asynq.Timeout(time.Minute))
 
@@ -185,10 +187,10 @@ func main() {
         log.Fatal("could not enqueue task: %v", err)
     }
 
-    // --------------------------------------------------------------------------
+    // ---------------------------------------------------------------------------
     // Example 4: Pass options to tune task processing behavior at enqueue time.
     //            Options passed at enqueue time override default ones, if any.
-    // --------------------------------------------------------------------------
+    // ---------------------------------------------------------------------------
 
     t = tasks.NewImageProcessingTask("some/blobstore/url", "other/blobstore/url")
     err = c.Enqueue(t, asynq.Queue("critical"), asynq.Timeout(30*time.Second))

asynq_test.go

@@ -5,7 +5,7 @@
 package asynq
 
 import (
-	"os"
+	"flag"
 	"sort"
 	"testing"
 
@@ -15,16 +15,28 @@ import (
 	"github.com/hibiken/asynq/internal/log"
 )
 
-// This file defines test helper functions used by
-// other test files.
+//============================================================================
+// This file defines helper functions and variables used in other test files.
+//============================================================================
 
-// redis used for package testing.
-const (
-	redisAddr = "localhost:6379"
-	redisDB   = 14
+// variables used for package testing.
+var (
+	redisAddr string
+	redisDB   int
+
+	testLogLevel = FatalLevel
 )
 
-var testLogger = log.NewLogger(os.Stderr)
+var testLogger *log.Logger
+
+func init() {
+	flag.StringVar(&redisAddr, "redis_addr", "localhost:6379", "redis address to use in testing")
+	flag.IntVar(&redisDB, "redis_db", 14, "redis db number to use in testing")
+	flag.Var(&testLogLevel, "loglevel", "log level to use in testing")
+
+	testLogger = log.NewLogger(nil)
+	testLogger.SetLevel(toInternalLogLevel(testLogLevel))
+}
 
 func setup(tb testing.TB) *redis.Client {
 	tb.Helper()

client.go

@@ -209,6 +209,11 @@ func (c *Client) EnqueueIn(d time.Duration, task *Task, opts ...Option) error {
 	return c.enqueueAt(time.Now().Add(d), task, opts...)
 }
 
+// Close closes the connection with redis server.
+func (c *Client) Close() error {
+	return c.rdb.Close()
+}
+
 func (c *Client) enqueueAt(t time.Time, task *Task, opts ...Option) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()

context.go

@@ -0,0 +1,85 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq
+
+import (
+	"context"
+	"time"
+
+	"github.com/hibiken/asynq/internal/base"
+)
+
+// A taskMetadata holds task scoped data to put in context.
+type taskMetadata struct {
+	id         string
+	maxRetry   int
+	retryCount int
+}
+
+// ctxKey type is unexported to prevent collisions with context keys defined in
+// other packages.
+type ctxKey int
+
+// metadataCtxKey is the context key for the task metadata.
+// Its value of zero is arbitrary.
+const metadataCtxKey ctxKey = 0
+
+// createContext returns a context and cancel function for a given task message.
+func createContext(msg *base.TaskMessage) (ctx context.Context, cancel context.CancelFunc) {
+	metadata := taskMetadata{
+		id:         msg.ID.String(),
+		maxRetry:   msg.Retry,
+		retryCount: msg.Retried,
+	}
+	ctx = context.WithValue(context.Background(), metadataCtxKey, metadata)
+	timeout, err := time.ParseDuration(msg.Timeout)
+	if err == nil && timeout != 0 {
+		ctx, cancel = context.WithTimeout(ctx, timeout)
+	}
+	deadline, err := time.Parse(time.RFC3339, msg.Deadline)
+	if err == nil && !deadline.IsZero() {
+		ctx, cancel = context.WithDeadline(ctx, deadline)
+	}
+	if cancel == nil {
+		ctx, cancel = context.WithCancel(ctx)
+	}
+	return ctx, cancel
+}
+
+// GetTaskID extracts a task ID from a context, if any.
+//
+// ID of a task is guaranteed to be unique.
+// ID of a task doesn't change if the task is being retried.
+func GetTaskID(ctx context.Context) (id string, ok bool) {
+	metadata, ok := ctx.Value(metadataCtxKey).(taskMetadata)
+	if !ok {
+		return "", false
+	}
+	return metadata.id, true
+}
+
+// GetRetryCount extracts retry count from a context, if any.
+//
+// Return value n indicates the number of times associated task has been
+// retried so far.
+func GetRetryCount(ctx context.Context) (n int, ok bool) {
+	metadata, ok := ctx.Value(metadataCtxKey).(taskMetadata)
+	if !ok {
+		return 0, false
+	}
+	return metadata.retryCount, true
+}
+
+// GetMaxRetry extracts maximum retry from a context, if any.
+//
+// Return value n indicates the maximum number of times the assoicated task
+// can be retried if ProcessTask returns a non-nil error.
+func GetMaxRetry(ctx context.Context) (n int, ok bool) {
+	metadata, ok := ctx.Value(metadataCtxKey).(taskMetadata)
+	if !ok {
+		return 0, false
+	}
+	return metadata.maxRetry, true
+}

context_test.go

@@ -0,0 +1,157 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package asynq
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/rs/xid"
+)
+
+func TestCreateContextWithTimeRestrictions(t *testing.T) {
+	var (
+		noTimeout  = time.Duration(0)
+		noDeadline = time.Time{}
+	)
+
+	tests := []struct {
+		desc         string
+		timeout      time.Duration
+		deadline     time.Time
+		wantDeadline time.Time
+	}{
+		{"only with timeout", 10 * time.Second, noDeadline, time.Now().Add(10 * time.Second)},
+		{"only with deadline", noTimeout, time.Now().Add(time.Hour), time.Now().Add(time.Hour)},
+		{"with timeout and deadline (timeout < deadline)", 10 * time.Second, time.Now().Add(time.Hour), time.Now().Add(10 * time.Second)},
+		{"with timeout and deadline (timeout > deadline)", 10 * time.Minute, time.Now().Add(30 * time.Second), time.Now().Add(30 * time.Second)},
+	}
+
+	for _, tc := range tests {
+		msg := &base.TaskMessage{
+			Type:     "something",
+			ID:       xid.New(),
+			Timeout:  tc.timeout.String(),
+			Deadline: tc.deadline.Format(time.RFC3339),
+		}
+
+		ctx, cancel := createContext(msg)
+
+		select {
+		case x := <-ctx.Done():
+			t.Errorf("%s: <-ctx.Done() == %v, want nothing (it should block)", tc.desc, x)
+		default:
+		}
+
+		got, ok := ctx.Deadline()
+		if !ok {
+			t.Errorf("%s: ctx.Deadline() returned false, want deadline to be set", tc.desc)
+		}
+		if !cmp.Equal(tc.wantDeadline, got, cmpopts.EquateApproxTime(time.Second)) {
+			t.Errorf("%s: ctx.Deadline() returned %v, want %v", tc.desc, got, tc.wantDeadline)
+		}
+
+		cancel()
+
+		select {
+		case <-ctx.Done():
+		default:
+			t.Errorf("ctx.Done() blocked, want it to be non-blocking")
+		}
+	}
+}
+
+func TestCreateContextWithoutTimeRestrictions(t *testing.T) {
+	msg := &base.TaskMessage{
+		Type:     "something",
+		ID:       xid.New(),
+		Timeout:  time.Duration(0).String(),        // zero value to indicate no timeout
+		Deadline: time.Time{}.Format(time.RFC3339), // zero value to indicate no deadline
+	}
+
+	ctx, cancel := createContext(msg)
+
+	select {
+	case x := <-ctx.Done():
+		t.Errorf("<-ctx.Done() == %v, want nothing (it should block)", x)
+	default:
+	}
+
+	_, ok := ctx.Deadline()
+	if ok {
+		t.Error("ctx.Deadline() returned true, want deadline to not be set")
+	}
+
+	cancel()
+
+	select {
+	case <-ctx.Done():
+	default:
+		t.Error("ctx.Done() blocked, want it to be non-blocking")
+	}
+}
+
+func TestGetTaskMetadataFromContext(t *testing.T) {
+	tests := []struct {
+		desc string
+		msg  *base.TaskMessage
+	}{
+		{"with zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 25, Retried: 0}},
+		{"with non-zero retried message", &base.TaskMessage{Type: "something", ID: xid.New(), Retry: 10, Retried: 5}},
+	}
+
+	for _, tc := range tests {
+		ctx, _ := createContext(tc.msg)
+
+		id, ok := GetTaskID(ctx)
+		if !ok {
+			t.Errorf("%s: GetTaskID(ctx) returned ok == false", tc.desc)
+		}
+		if ok && id != tc.msg.ID.String() {
+			t.Errorf("%s: GetTaskID(ctx) returned id == %q, want %q", tc.desc, id, tc.msg.ID.String())
+		}
+
+		retried, ok := GetRetryCount(ctx)
+		if !ok {
+			t.Errorf("%s: GetRetryCount(ctx) returned ok == false", tc.desc)
+		}
+		if ok && retried != tc.msg.Retried {
+			t.Errorf("%s: GetRetryCount(ctx) returned n == %d want %d", tc.desc, retried, tc.msg.Retried)
+		}
+
+		maxRetry, ok := GetMaxRetry(ctx)
+		if !ok {
+			t.Errorf("%s: GetMaxRetry(ctx) returned ok == false", tc.desc)
+		}
+		if ok && maxRetry != tc.msg.Retry {
+			t.Errorf("%s: GetMaxRetry(ctx) returned n == %d want %d", tc.desc, maxRetry, tc.msg.Retry)
+		}
+	}
+}
+
+func TestGetTaskMetadataFromContextError(t *testing.T) {
+	tests := []struct {
+		desc string
+		ctx  context.Context
+	}{
+		{"with background context", context.Background()},
+	}
+
+	for _, tc := range tests {
+		if _, ok := GetTaskID(tc.ctx); ok {
+			t.Errorf("%s: GetTaskID(ctx) returned ok == true", tc.desc)
+		}
+		if _, ok := GetRetryCount(tc.ctx); ok {
+			t.Errorf("%s: GetRetryCount(ctx) returned ok == true", tc.desc)
+		}
+		if _, ok := GetMaxRetry(tc.ctx); ok {
+			t.Errorf("%s: GetMaxRetry(ctx) returned ok == true", tc.desc)
+		}
+	}
+}

heartbeat.go

@@ -9,12 +9,13 @@ import (
 	"time"
 
 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 )
 
 // heartbeater is responsible for writing process info to redis periodically to
 // indicate that the background worker process is up.
 type heartbeater struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker
 
 	ss *base.ServerState
@@ -26,18 +27,25 @@ type heartbeater struct {
 	interval time.Duration
 }
 
-func newHeartbeater(l Logger, b base.Broker, ss *base.ServerState, interval time.Duration) *heartbeater {
+type heartbeaterParams struct {
+	logger      *log.Logger
+	broker      base.Broker
+	serverState *base.ServerState
+	interval    time.Duration
+}
+
+func newHeartbeater(params heartbeaterParams) *heartbeater {
 	return &heartbeater{
-		logger:   l,
-		broker:   b,
-		ss:       ss,
+		logger:   params.logger,
+		broker:   params.broker,
+		ss:       params.serverState,
 		done:     make(chan struct{}),
-		interval: interval,
+		interval: params.interval,
 	}
 }
 
 func (h *heartbeater) terminate() {
-	h.logger.Info("Heartbeater shutting down...")
+	h.logger.Debug("Heartbeater shutting down...")
 	// Signal the heartbeater goroutine to stop.
 	h.done <- struct{}{}
 }
@@ -53,7 +61,7 @@ func (h *heartbeater) start(wg *sync.WaitGroup) {
 			select {
 			case <-h.done:
 				h.broker.ClearServerState(h.ss)
-				h.logger.Info("Heartbeater done")
+				h.logger.Debug("Heartbeater done")
 				return
 			case <-time.After(h.interval):
 				h.beat()
@@ -67,6 +75,6 @@ func (h *heartbeater) beat() {
 	// and short enough to expire quickly once the process is shut down or killed.
 	err := h.broker.WriteServerState(h.ss, h.interval*2)
 	if err != nil {
-		h.logger.Error("could not write heartbeat data: %v", err)
+		h.logger.Errorf("could not write heartbeat data: %v", err)
 	}
 }

heartbeat_test.go

@@ -38,7 +38,12 @@ func TestHeartbeater(t *testing.T) {
 		h.FlushDB(t, r)
 
 		state := base.NewServerState(tc.host, tc.pid, tc.concurrency, tc.queues, false)
-		hb := newHeartbeater(testLogger, rdbClient, state, tc.interval)
+		hb := newHeartbeater(heartbeaterParams{
+			logger:      testLogger,
+			broker:      rdbClient,
+			serverState: state,
+			interval:    tc.interval,
+		})
 
 		var wg sync.WaitGroup
 		hb.start(&wg)
@@ -115,7 +120,12 @@ func TestHeartbeaterWithRedisDown(t *testing.T) {
 	r := rdb.NewRDB(setup(t))
 	testBroker := testbroker.NewTestBroker(r)
 	ss := base.NewServerState("localhost", 1234, 10, map[string]int{"default": 1}, false)
-	hb := newHeartbeater(testLogger, testBroker, ss, time.Second)
+	hb := newHeartbeater(heartbeaterParams{
+		logger:      testLogger,
+		broker:      testBroker,
+		serverState: ss,
+		interval:    time.Second,
+	})
 
 	testBroker.Sleep()
 	var wg sync.WaitGroup

internal/log/log.go

@@ -6,52 +6,210 @@
 package log
 
 import (
+	"fmt"
 	"io"
 	stdlog "log"
 	"os"
+	"sync"
 )
 
-// NewLogger creates and returns a new instance of Logger.
-func NewLogger(out io.Writer) *Logger {
-	return &Logger{
-		stdlog.New(out, "", stdlog.Ldate|stdlog.Ltime|stdlog.Lmicroseconds|stdlog.LUTC),
-	}
+// Base supports logging at various log levels.
+type Base interface {
+	// Debug logs a message at Debug level.
+	Debug(args ...interface{})
+
+	// Info logs a message at Info level.
+	Info(args ...interface{})
+
+	// Warn logs a message at Warning level.
+	Warn(args ...interface{})
+
+	// Error logs a message at Error level.
+	Error(args ...interface{})
+
+	// Fatal logs a message at Fatal level
+	// and process will exit with status set to 1.
+	Fatal(args ...interface{})
 }
 
-// Logger is a wrapper object around log.Logger from the standard library.
+// baseLogger is a wrapper object around log.Logger from the standard library.
 // It supports logging at various log levels.
-type Logger struct {
+type baseLogger struct {
 	*stdlog.Logger
 }
 
 // Debug logs a message at Debug level.
-func (l *Logger) Debug(format string, args ...interface{}) {
-	format = "DEBUG: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Debug(args ...interface{}) {
+	l.prefixPrint("DEBUG: ", args...)
 }
 
 // Info logs a message at Info level.
-func (l *Logger) Info(format string, args ...interface{}) {
-	format = "INFO: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Info(args ...interface{}) {
+	l.prefixPrint("INFO: ", args...)
 }
 
 // Warn logs a message at Warning level.
-func (l *Logger) Warn(format string, args ...interface{}) {
-	format = "WARN: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Warn(args ...interface{}) {
+	l.prefixPrint("WARN: ", args...)
 }
 
 // Error logs a message at Error level.
-func (l *Logger) Error(format string, args ...interface{}) {
-	format = "ERROR: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Error(args ...interface{}) {
+	l.prefixPrint("ERROR: ", args...)
 }
 
 // Fatal logs a message at Fatal level
 // and process will exit with status set to 1.
-func (l *Logger) Fatal(format string, args ...interface{}) {
-	format = "FATAL: " + format
-	l.Printf(format, args...)
+func (l *baseLogger) Fatal(args ...interface{}) {
+	l.prefixPrint("FATAL: ", args...)
 	os.Exit(1)
 }
+
+func (l *baseLogger) prefixPrint(prefix string, args ...interface{}) {
+	args = append([]interface{}{prefix}, args...)
+	l.Print(args...)
+}
+
+// newBase creates and returns a new instance of baseLogger.
+func newBase(out io.Writer) *baseLogger {
+	prefix := fmt.Sprintf("asynq: pid=%d ", os.Getpid())
+	return &baseLogger{
+		stdlog.New(out, prefix, stdlog.Ldate|stdlog.Ltime|stdlog.Lmicroseconds|stdlog.LUTC),
+	}
+}
+
+// NewLogger creates and returns a new instance of Logger.
+// Log level is set to DebugLevel by default.
+func NewLogger(base Base) *Logger {
+	if base == nil {
+		base = newBase(os.Stderr)
+	}
+	return &Logger{base: base, level: DebugLevel}
+}
+
+// Logger logs message to io.Writer at various log levels.
+type Logger struct {
+	base Base
+
+	mu sync.Mutex
+	// Minimum log level for this logger.
+	// Message with level lower than this level won't be outputted.
+	level Level
+}
+
+// Level represents a log level.
+type Level int32
+
+const (
+	// DebugLevel is the lowest level of logging.
+	// Debug logs are intended for debugging and development purposes.
+	DebugLevel Level = iota
+
+	// InfoLevel is used for general informational log messages.
+	InfoLevel
+
+	// WarnLevel is used for undesired but relatively expected events,
+	// which may indicate a problem.
+	WarnLevel
+
+	// ErrorLevel is used for undesired and unexpected events that
+	// the program can recover from.
+	ErrorLevel
+
+	// FatalLevel is used for undesired and unexpected events that
+	// the program cannot recover from.
+	FatalLevel
+)
+
+// String is part of the fmt.Stringer interface.
+//
+// Used for testing and debugging purposes.
+func (l Level) String() string {
+	switch l {
+	case DebugLevel:
+		return "debug"
+	case InfoLevel:
+		return "info"
+	case WarnLevel:
+		return "warning"
+	case ErrorLevel:
+		return "error"
+	case FatalLevel:
+		return "fatal"
+	default:
+		return "unknown"
+	}
+}
+
+// canLogAt reports whether logger can log at level v.
+func (l *Logger) canLogAt(v Level) bool {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return v >= l.level
+}
+
+func (l *Logger) Debug(args ...interface{}) {
+	if !l.canLogAt(DebugLevel) {
+		return
+	}
+	l.base.Debug(args...)
+}
+
+func (l *Logger) Info(args ...interface{}) {
+	if !l.canLogAt(InfoLevel) {
+		return
+	}
+	l.base.Info(args...)
+}
+
+func (l *Logger) Warn(args ...interface{}) {
+	if !l.canLogAt(WarnLevel) {
+		return
+	}
+	l.base.Warn(args...)
+}
+
+func (l *Logger) Error(args ...interface{}) {
+	if !l.canLogAt(ErrorLevel) {
+		return
+	}
+	l.base.Error(args...)
+}
+
+func (l *Logger) Fatal(args ...interface{}) {
+	if !l.canLogAt(FatalLevel) {
+		return
+	}
+	l.base.Fatal(args...)
+}
+
+func (l *Logger) Debugf(format string, args ...interface{}) {
+	l.Debug(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Infof(format string, args ...interface{}) {
+	l.Info(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Warnf(format string, args ...interface{}) {
+	l.Warn(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Errorf(format string, args ...interface{}) {
+	l.Error(fmt.Sprintf(format, args...))
+}
+
+func (l *Logger) Fatalf(format string, args ...interface{}) {
+	l.Fatal(fmt.Sprintf(format, args...))
+}
+
+// SetLevel sets the logger level.
+// It panics if v is less than DebugLevel or greater than FatalLevel.
+func (l *Logger) SetLevel(v Level) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if v < DebugLevel || v > FatalLevel {
+		panic("log: invalid log level")
+	}
+	l.level = v
+}

internal/log/log_test.go

@@ -13,6 +13,7 @@ import (
 
 // regexp for timestamps
 const (
+	rgxPID          = `[0-9]+`
 	rgxdate         = `[0-9][0-9][0-9][0-9]/[0-9][0-9]/[0-9][0-9]`
 	rgxtime         = `[0-9][0-9]:[0-9][0-9]:[0-9][0-9]`
 	rgxmicroseconds = `\.[0-9][0-9][0-9][0-9][0-9][0-9]`
@@ -29,18 +30,20 @@ func TestLoggerDebug(t *testing.T) {
 		{
 			desc:    "without trailing newline, logger adds newline",
 			message: "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s DEBUG: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s DEBUG: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
 			desc:    "with trailing newline, logger preserves newline",
 			message: "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s DEBUG: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s DEBUG: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}
 
 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))
 
 		logger.Debug(tc.message)
 
@@ -50,7 +53,7 @@ func TestLoggerDebug(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Debug(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
@@ -61,18 +64,20 @@ func TestLoggerInfo(t *testing.T) {
 		{
 			desc:    "without trailing newline, logger adds newline",
 			message: "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s INFO: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s INFO: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
 			desc:    "with trailing newline, logger preserves newline",
 			message: "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s INFO: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s INFO: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}
 
 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))
 
 		logger.Info(tc.message)
 
@@ -82,7 +87,7 @@ func TestLoggerInfo(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Info(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
@@ -93,18 +98,20 @@ func TestLoggerWarn(t *testing.T) {
 		{
 			desc:    "without trailing newline, logger adds newline",
 			message: "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s WARN: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s WARN: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
 			desc:    "with trailing newline, logger preserves newline",
 			message: "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s WARN: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s WARN: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}
 
 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))
 
 		logger.Warn(tc.message)
 
@@ -114,7 +121,7 @@ func TestLoggerWarn(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Warn(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
@@ -125,18 +132,20 @@ func TestLoggerError(t *testing.T) {
 		{
 			desc:    "without trailing newline, logger adds newline",
 			message: "hello, world!",
-			wantPattern: fmt.Sprintf("^%s %s%s ERROR: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s ERROR: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 		{
 			desc:    "with trailing newline, logger preserves newline",
 			message: "hello, world!\n",
-			wantPattern: fmt.Sprintf("^%s %s%s ERROR: hello, world!\n$", rgxdate, rgxtime, rgxmicroseconds),
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s ERROR: hello, world!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
 		},
 	}
 
 	for _, tc := range tests {
 		var buf bytes.Buffer
-		logger := NewLogger(&buf)
+		logger := NewLogger(newBase(&buf))
 
 		logger.Error(tc.message)
 
@@ -146,8 +155,234 @@ func TestLoggerError(t *testing.T) {
 			t.Fatal("pattern did not compile:", err)
 		}
 		if !matched {
-			t.Errorf("logger.info(%q) outputted %q, should match pattern %q",
+			t.Errorf("logger.Error(%q) outputted %q, should match pattern %q",
 				tc.message, got, tc.wantPattern)
 		}
 	}
 }
+
+type formatTester struct {
+	desc        string
+	format      string
+	args        []interface{}
+	wantPattern string // regexp that log output must match
+}
+
+func TestLoggerDebugf(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with DEBUG prefix",
+			format: "hello, %s!",
+			args:   []interface{}{"Gopher"},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s DEBUG: hello, Gopher!\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Debugf(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Debugf(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerInfof(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with INFO prefix",
+			format: "%d,%d,%d",
+			args:   []interface{}{1, 2, 3},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s INFO: 1,2,3\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Infof(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Infof(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerWarnf(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with WARN prefix",
+			format: "hello, %s",
+			args:   []interface{}{"Gophers"},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s WARN: hello, Gophers\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Warnf(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Warnf(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerErrorf(t *testing.T) {
+	tests := []formatTester{
+		{
+			desc:   "Formats message with ERROR prefix",
+			format: "hello, %s",
+			args:   []interface{}{"Gophers"},
+			wantPattern: fmt.Sprintf("^asynq: pid=%s %s %s%s ERROR: hello, Gophers\n$",
+				rgxPID, rgxdate, rgxtime, rgxmicroseconds),
+		},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+
+		logger.Errorf(tc.format, tc.args...)
+
+		got := buf.String()
+		matched, err := regexp.MatchString(tc.wantPattern, got)
+		if err != nil {
+			t.Fatal("pattern did not compile:", err)
+		}
+		if !matched {
+			t.Errorf("logger.Errorf(%q, %v) outputted %q, should match pattern %q",
+				tc.format, tc.args, got, tc.wantPattern)
+		}
+	}
+}
+
+func TestLoggerWithLowerLevels(t *testing.T) {
+	// Logger should not log messages at a level
+	// lower than the specified level.
+	tests := []struct {
+		level Level
+		op    string
+	}{
+		// with level one above
+		{InfoLevel, "Debug"},
+		{InfoLevel, "Debugf"},
+		{WarnLevel, "Info"},
+		{WarnLevel, "Infof"},
+		{ErrorLevel, "Warn"},
+		{ErrorLevel, "Warnf"},
+		{FatalLevel, "Error"},
+		{FatalLevel, "Errorf"},
+		// with skip level
+		{WarnLevel, "Debug"},
+		{ErrorLevel, "Infof"},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+		logger.SetLevel(tc.level)
+
+		switch tc.op {
+		case "Debug":
+			logger.Debug("hello")
+		case "Debugf":
+			logger.Debugf("hello, %s", "world")
+		case "Info":
+			logger.Info("hello")
+		case "Infof":
+			logger.Infof("hello, %s", "world")
+		case "Warn":
+			logger.Warn("hello")
+		case "Warnf":
+			logger.Warnf("hello, %s", "world")
+		case "Error":
+			logger.Error("hello")
+		case "Errorf":
+			logger.Errorf("hello, %s", "world")
+		default:
+			t.Fatalf("unexpected op: %q", tc.op)
+		}
+
+		if buf.String() != "" {
+			t.Errorf("logger.%s outputted log message when level is set to %v", tc.op, tc.level)
+		}
+	}
+}
+
+func TestLoggerWithSameOrHigherLevels(t *testing.T) {
+	// Logger should log messages at a level
+	// same as or higher than the specified level.
+	tests := []struct {
+		level Level
+		op    string
+	}{
+		// same level
+		{DebugLevel, "Debug"},
+		{InfoLevel, "Infof"},
+		{WarnLevel, "Warn"},
+		{ErrorLevel, "Errorf"},
+		// higher level
+		{DebugLevel, "Info"},
+		{InfoLevel, "Warnf"},
+		{WarnLevel, "Error"},
+	}
+
+	for _, tc := range tests {
+		var buf bytes.Buffer
+		logger := NewLogger(newBase(&buf))
+		logger.SetLevel(tc.level)
+
+		switch tc.op {
+		case "Debug":
+			logger.Debug("hello")
+		case "Debugf":
+			logger.Debugf("hello, %s", "world")
+		case "Info":
+			logger.Info("hello")
+		case "Infof":
+			logger.Infof("hello, %s", "world")
+		case "Warn":
+			logger.Warn("hello")
+		case "Warnf":
+			logger.Warnf("hello, %s", "world")
+		case "Error":
+			logger.Error("hello")
+		case "Errorf":
+			logger.Errorf("hello, %s", "world")
+		default:
+			t.Fatalf("unexpected op: %q", tc.op)
+		}
+
+		if buf.String() == "" {
+			t.Errorf("logger.%s did not output log message when level is set to %v", tc.op, tc.level)
+		}
+	}
+}

processor.go

@@ -13,12 +13,13 @@ import (
 	"time"
 
 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 	"github.com/hibiken/asynq/internal/rdb"
 	"golang.org/x/time/rate"
 )
 
 type processor struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker
 
 	ss *base.ServerState
@@ -63,8 +64,8 @@ type processor struct {
 
 type retryDelayFunc func(n int, err error, task *Task) time.Duration
 
-type newProcessorParams struct {
-	logger          Logger
+type processorParams struct {
+	logger          *log.Logger
 	broker          base.Broker
 	ss              *base.ServerState
 	retryDelayFunc  retryDelayFunc
@@ -75,7 +76,7 @@ type newProcessorParams struct {
 }
 
 // newProcessor constructs a new processor.
-func newProcessor(params newProcessorParams) *processor {
+func newProcessor(params processorParams) *processor {
 	info := params.ss.GetInfo()
 	qcfg := normalizeQueueCfg(info.Queues)
 	orderedQueues := []string(nil)
@@ -105,7 +106,7 @@ func newProcessor(params newProcessorParams) *processor {
 // It's safe to call this method multiple times.
 func (p *processor) stop() {
 	p.once.Do(func() {
-		p.logger.Info("Processor shutting down...")
+		p.logger.Debug("Processor shutting down...")
 		// Unblock if processor is waiting for sema token.
 		close(p.abort)
 		// Signal the processor goroutine to stop processing tasks
@@ -144,7 +145,7 @@ func (p *processor) start(wg *sync.WaitGroup) {
 		for {
 			select {
 			case <-p.done:
-				p.logger.Info("Processor done")
+				p.logger.Debug("Processor done")
 				return
 			default:
 				p.exec()
@@ -158,19 +159,20 @@ func (p *processor) start(wg *sync.WaitGroup) {
 func (p *processor) exec() {
 	qnames := p.queues()
 	msg, err := p.broker.Dequeue(qnames...)
-	if err == rdb.ErrNoProcessableTask { // TODO: Need to decouple this error from rdb to support other brokers
+	switch {
+	case err == rdb.ErrNoProcessableTask:
 		// queues are empty, this is a normal behavior.
-		if len(p.queueConfig) > 1 {
+		if len(qnames) > 1 {
 			// sleep to avoid slamming redis and let scheduler move tasks into queues.
 			// Note: With multiple queues, we are not using blocking pop operation and
 			// polling queues instead. This adds significant load to redis.
 			time.Sleep(time.Second)
 		}
+		p.logger.Debug("All queues are empty")
 		return
-	}
-	if err != nil {
+	case err != nil:
 		if p.errLogLimiter.Allow() {
-			p.logger.Error("Dequeue error: %v", err)
+			p.logger.Errorf("Dequeue error: %v", err)
 		}
 		return
 	}
@@ -185,7 +187,7 @@ func (p *processor) exec() {
 		go func() {
 			defer func() {
 				p.ss.DeleteWorkerStats(msg)
-				<-p.sema /* release token */
+				<-p.sema // release token
 			}()
 
 			ctx, cancel := createContext(msg)
@@ -202,7 +204,7 @@ func (p *processor) exec() {
 			select {
 			case <-p.quit:
 				// time is up, quit this worker goroutine.
-				p.logger.Warn("Quitting worker. task id=%s", msg.ID)
+				p.logger.Warnf("Quitting worker. task id=%s", msg.ID)
 				return
 			case resErr := <-resCh:
 				// Note: One of three things should happen.
@@ -231,17 +233,17 @@ func (p *processor) exec() {
 func (p *processor) restore() {
 	n, err := p.broker.RequeueAll()
 	if err != nil {
-		p.logger.Error("Could not restore unfinished tasks: %v", err)
+		p.logger.Errorf("Could not restore unfinished tasks: %v", err)
 	}
 	if n > 0 {
-		p.logger.Info("Restored %d unfinished tasks back to queue", n)
+		p.logger.Infof("Restored %d unfinished tasks back to queue", n)
 	}
 }
 
 func (p *processor) requeue(msg *base.TaskMessage) {
 	err := p.broker.Requeue(msg)
 	if err != nil {
-		p.logger.Error("Could not push task id=%s back to queue: %v", msg.ID, err)
+		p.logger.Errorf("Could not push task id=%s back to queue: %v", msg.ID, err)
 	}
 }
 
@@ -249,7 +251,7 @@ func (p *processor) markAsDone(msg *base.TaskMessage) {
 	err := p.broker.Done(msg)
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not remove task id=%s from %q", msg.ID, base.InProgressQueue)
-		p.logger.Warn("%s; Will retry syncing", errMsg)
+		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Done(msg)
@@ -265,7 +267,7 @@ func (p *processor) retry(msg *base.TaskMessage, e error) {
 	err := p.broker.Retry(msg, retryAt, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.RetryQueue)
-		p.logger.Warn("%s; Will retry syncing", errMsg)
+		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Retry(msg, retryAt, e.Error())
@@ -276,11 +278,11 @@ func (p *processor) retry(msg *base.TaskMessage, e error) {
 }
 
 func (p *processor) kill(msg *base.TaskMessage, e error) {
-	p.logger.Warn("Retry exhausted for task id=%s", msg.ID)
+	p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
 	err := p.broker.Kill(msg, e.Error())
 	if err != nil {
 		errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.DeadQueue)
-		p.logger.Warn("%s; Will retry syncing", errMsg)
+		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
 				return p.broker.Kill(msg, e.Error())
@@ -308,7 +310,7 @@ func (p *processor) queues() []string {
 	}
 	var names []string
 	for qname, priority := range p.queueConfig {
-		for i := 0; i < int(priority); i++ {
+		for i := 0; i < priority; i++ {
 			names = append(names, qname)
 		}
 	}
@@ -403,20 +405,3 @@ func gcd(xs ...int) int {
 	}
 	return res
 }
-
-// createContext returns a context and cancel function for a given task message.
-func createContext(msg *base.TaskMessage) (ctx context.Context, cancel context.CancelFunc) {
-	ctx = context.Background()
-	timeout, err := time.ParseDuration(msg.Timeout)
-	if err == nil && timeout != 0 {
-		ctx, cancel = context.WithTimeout(ctx, timeout)
-	}
-	deadline, err := time.Parse(time.RFC3339, msg.Deadline)
-	if err == nil && !deadline.IsZero() {
-		ctx, cancel = context.WithDeadline(ctx, deadline)
-	}
-	if cancel == nil {
-		ctx, cancel = context.WithCancel(ctx)
-	}
-	return ctx, cancel
-}

processor_test.go

@@ -17,7 +17,6 @@ import (
 	h "github.com/hibiken/asynq/internal/asynqtest"
 	"github.com/hibiken/asynq/internal/base"
 	"github.com/hibiken/asynq/internal/rdb"
-	"github.com/rs/xid"
 )
 
 func TestProcessorSuccess(t *testing.T) {
@@ -65,7 +64,7 @@ func TestProcessorSuccess(t *testing.T) {
 			return nil
 		}
 		ss := base.NewServerState("localhost", 1234, 10, defaultQueueConfig, false)
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          rdbClient,
 			ss:              ss,
@@ -170,7 +169,7 @@ func TestProcessorRetry(t *testing.T) {
 			n++
 		}
 		ss := base.NewServerState("localhost", 1234, 10, defaultQueueConfig, false)
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          rdbClient,
 			ss:              ss,
@@ -243,7 +242,7 @@ func TestProcessorQueues(t *testing.T) {
 
 	for _, tc := range tests {
 		ss := base.NewServerState("localhost", 1234, 10, tc.queueCfg, false)
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          nil,
 			ss:              ss,
@@ -319,7 +318,7 @@ func TestProcessorWithStrictPriority(t *testing.T) {
 		}
 		// Note: Set concurrency to 1 to make sure tasks are processed one at a time.
 		ss := base.NewServerState("localhost", 1234, 1 /* concurrency */, queueCfg, true /*strict*/)
-		p := newProcessor(newProcessorParams{
+		p := newProcessor(processorParams{
 			logger:          testLogger,
 			broker:          rdbClient,
 			ss:              ss,
@@ -391,88 +390,6 @@ func TestPerform(t *testing.T) {
 	}
 }
 
-func TestCreateContextWithTimeRestrictions(t *testing.T) {
-	var (
-		noTimeout  = time.Duration(0)
-		noDeadline = time.Time{}
-	)
-
-	tests := []struct {
-		desc         string
-		timeout      time.Duration
-		deadline     time.Time
-		wantDeadline time.Time
-	}{
-		{"only with timeout", 10 * time.Second, noDeadline, time.Now().Add(10 * time.Second)},
-		{"only with deadline", noTimeout, time.Now().Add(time.Hour), time.Now().Add(time.Hour)},
-		{"with timeout and deadline (timeout < deadline)", 10 * time.Second, time.Now().Add(time.Hour), time.Now().Add(10 * time.Second)},
-		{"with timeout and deadline (timeout > deadline)", 10 * time.Minute, time.Now().Add(30 * time.Second), time.Now().Add(30 * time.Second)},
-	}
-
-	for _, tc := range tests {
-		msg := &base.TaskMessage{
-			Type:     "something",
-			ID:       xid.New(),
-			Timeout:  tc.timeout.String(),
-			Deadline: tc.deadline.Format(time.RFC3339),
-		}
-
-		ctx, cancel := createContext(msg)
-
-		select {
-		case x := <-ctx.Done():
-			t.Errorf("%s: <-ctx.Done() == %v, want nothing (it should block)", tc.desc, x)
-		default:
-		}
-
-		got, ok := ctx.Deadline()
-		if !ok {
-			t.Errorf("%s: ctx.Deadline() returned false, want deadline to be set", tc.desc)
-		}
-		if !cmp.Equal(tc.wantDeadline, got, cmpopts.EquateApproxTime(time.Second)) {
-			t.Errorf("%s: ctx.Deadline() returned %v, want %v", tc.desc, got, tc.wantDeadline)
-		}
-
-		cancel()
-
-		select {
-		case <-ctx.Done():
-		default:
-			t.Errorf("ctx.Done() blocked, want it to be non-blocking")
-		}
-	}
-}
-
-func TestCreateContextWithoutTimeRestrictions(t *testing.T) {
-	msg := &base.TaskMessage{
-		Type:     "something",
-		ID:       xid.New(),
-		Timeout:  time.Duration(0).String(),        // zero value to indicate no timeout
-		Deadline: time.Time{}.Format(time.RFC3339), // zero value to indicate no deadline
-	}
-
-	ctx, cancel := createContext(msg)
-
-	select {
-	case x := <-ctx.Done():
-		t.Errorf("<-ctx.Done() == %v, want nothing (it should block)", x)
-	default:
-	}
-
-	_, ok := ctx.Deadline()
-	if ok {
-		t.Error("ctx.Deadline() returned true, want deadline to not be set")
-	}
-
-	cancel()
-
-	select {
-	case <-ctx.Done():
-	default:
-		t.Error("ctx.Done() blocked, want it to be non-blocking")
-	}
-}
-
 func TestGCD(t *testing.T) {
 	tests := []struct {
 		input []int

scheduler.go

@@ -9,10 +9,11 @@ import (
 	"time"
 
 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 )
 
 type scheduler struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker
 
 	// channel to communicate back to the long running "scheduler" goroutine.
@@ -25,22 +26,29 @@ type scheduler struct {
 	qnames []string
 }
 
-func newScheduler(l Logger, b base.Broker, avgInterval time.Duration, qcfg map[string]int) *scheduler {
+type schedulerParams struct {
+	logger   *log.Logger
+	broker   base.Broker
+	interval time.Duration
+	queues   map[string]int
+}
+
+func newScheduler(params schedulerParams) *scheduler {
 	var qnames []string
-	for q := range qcfg {
+	for q := range params.queues {
 		qnames = append(qnames, q)
 	}
 	return &scheduler{
-		logger:      l,
-		broker:      b,
+		logger:      params.logger,
+		broker:      params.broker,
 		done:        make(chan struct{}),
-		avgInterval: avgInterval,
+		avgInterval: params.interval,
 		qnames:      qnames,
 	}
 }
 
 func (s *scheduler) terminate() {
-	s.logger.Info("Scheduler shutting down...")
+	s.logger.Debug("Scheduler shutting down...")
 	// Signal the scheduler goroutine to stop polling.
 	s.done <- struct{}{}
 }
@@ -53,7 +61,7 @@ func (s *scheduler) start(wg *sync.WaitGroup) {
 		for {
 			select {
 			case <-s.done:
-				s.logger.Info("Scheduler done")
+				s.logger.Debug("Scheduler done")
 				return
 			case <-time.After(s.avgInterval):
 				s.exec()
@@ -64,6 +72,6 @@ func (s *scheduler) start(wg *sync.WaitGroup) {
 
 func (s *scheduler) exec() {
 	if err := s.broker.CheckAndEnqueue(s.qnames...); err != nil {
-		s.logger.Error("Could not enqueue scheduled tasks: %v", err)
+		s.logger.Errorf("Could not enqueue scheduled tasks: %v", err)
 	}
 }

scheduler_test.go

@@ -19,7 +19,12 @@ func TestScheduler(t *testing.T) {
 	r := setup(t)
 	rdbClient := rdb.NewRDB(r)
 	const pollInterval = time.Second
-	s := newScheduler(testLogger, rdbClient, pollInterval, defaultQueueConfig)
+	s := newScheduler(schedulerParams{
+		logger:   testLogger,
+		broker:   rdbClient,
+		interval: pollInterval,
+		queues:   defaultQueueConfig,
+	})
 	t1 := h.NewTaskMessage("gen_thumbnail", nil)
 	t2 := h.NewTaskMessage("send_email", nil)
 	t3 := h.NewTaskMessage("reindex", nil)

servemux.go

@@ -15,7 +15,7 @@ import (
 // ServeMux is a multiplexer for asynchronous tasks.
 // It matches the type of each task against a list of registered patterns
 // and calls the handler for the pattern that most closely matches the
-// taks's type name.
+// task's type name.
 //
 // Longer patterns take precedence over shorter ones, so that if there are
 // handlers registered for both "images" and "images:thumbnails",

server.go

@@ -12,6 +12,7 @@ import (
 	"math/rand"
 	"os"
 	"runtime"
+	"strings"
 	"sync"
 	"time"
 
@@ -35,7 +36,7 @@ import (
 type Server struct {
 	ss *base.ServerState
 
-	logger Logger
+	logger *log.Logger
 
 	broker base.Broker
 
@@ -112,6 +113,11 @@ type Config struct {
 	// If unset, default logger is used.
 	Logger Logger
 
+	// LogLevel specifies the minimum log level to enable.
+	//
+	// If unset, InfoLevel is used by default.
+	LogLevel LogLevel
+
 	// ShutdownTimeout specifies the duration to wait to let workers finish their tasks
 	// before forcing them to abort when stopping the server.
 	//
@@ -133,23 +139,104 @@ func (fn ErrorHandlerFunc) HandleError(task *Task, err error, retried, maxRetry
 	fn(task, err, retried, maxRetry)
 }
 
-// Logger implements logging with various log levels.
+// Logger supports logging at various log levels.
 type Logger interface {
 	// Debug logs a message at Debug level.
-	Debug(format string, args ...interface{})
+	Debug(args ...interface{})
 
 	// Info logs a message at Info level.
-	Info(format string, args ...interface{})
+	Info(args ...interface{})
 
 	// Warn logs a message at Warning level.
-	Warn(format string, args ...interface{})
+	Warn(args ...interface{})
 
 	// Error logs a message at Error level.
-	Error(format string, args ...interface{})
+	Error(args ...interface{})
 
 	// Fatal logs a message at Fatal level
 	// and process will exit with status set to 1.
-	Fatal(format string, args ...interface{})
+	Fatal(args ...interface{})
+}
+
+// LogLevel represents logging level.
+//
+// It satisfies flag.Value interface.
+type LogLevel int32
+
+const (
+	// Note: reserving value zero to differentiate unspecified case.
+	level_unspecified LogLevel = iota
+
+	// DebugLevel is the lowest level of logging.
+	// Debug logs are intended for debugging and development purposes.
+	DebugLevel
+
+	// InfoLevel is used for general informational log messages.
+	InfoLevel
+
+	// WarnLevel is used for undesired but relatively expected events,
+	// which may indicate a problem.
+	WarnLevel
+
+	// ErrorLevel is used for undesired and unexpected events that
+	// the program can recover from.
+	ErrorLevel
+
+	// FatalLevel is used for undesired and unexpected events that
+	// the program cannot recover from.
+	FatalLevel
+)
+
+// String is part of the flag.Value interface.
+func (l *LogLevel) String() string {
+	switch *l {
+	case DebugLevel:
+		return "debug"
+	case InfoLevel:
+		return "info"
+	case WarnLevel:
+		return "warn"
+	case ErrorLevel:
+		return "error"
+	case FatalLevel:
+		return "fatal"
+	}
+	panic(fmt.Sprintf("asynq: unexpected log level: %v", *l))
+}
+
+// Set is part of the flag.Value interface.
+func (l *LogLevel) Set(val string) error {
+	switch strings.ToLower(val) {
+	case "debug":
+		*l = DebugLevel
+	case "info":
+		*l = InfoLevel
+	case "warn", "warning":
+		*l = WarnLevel
+	case "error":
+		*l = ErrorLevel
+	case "fatal":
+		*l = FatalLevel
+	default:
+		return fmt.Errorf("asynq: unsupported log level %q", val)
+	}
+	return nil
+}
+
+func toInternalLogLevel(l LogLevel) log.Level {
+	switch l {
+	case DebugLevel:
+		return log.DebugLevel
+	case InfoLevel:
+		return log.InfoLevel
+	case WarnLevel:
+		return log.WarnLevel
+	case ErrorLevel:
+		return log.ErrorLevel
+	case FatalLevel:
+		return log.FatalLevel
+	}
+	panic(fmt.Sprintf("asynq: unexpected log level: %v", l))
 }
 
 // Formula taken from https://github.com/mperham/sidekiq.
@@ -185,14 +272,16 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 	if len(queues) == 0 {
 		queues = defaultQueueConfig
 	}
-	logger := cfg.Logger
-	if logger == nil {
-		logger = log.NewLogger(os.Stderr)
-	}
 	shutdownTimeout := cfg.ShutdownTimeout
 	if shutdownTimeout == 0 {
 		shutdownTimeout = defaultShutdownTimeout
 	}
+	logger := log.NewLogger(cfg.Logger)
+	loglevel := cfg.LogLevel
+	if loglevel == level_unspecified {
+		loglevel = InfoLevel
+	}
+	logger.SetLevel(toInternalLogLevel(loglevel))
 
 	host, err := os.Hostname()
 	if err != nil {
@@ -204,11 +293,30 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 	ss := base.NewServerState(host, pid, n, queues, cfg.StrictPriority)
 	syncCh := make(chan *syncRequest)
 	cancels := base.NewCancelations()
-	syncer := newSyncer(logger, syncCh, 5*time.Second)
-	heartbeater := newHeartbeater(logger, rdb, ss, 5*time.Second)
-	scheduler := newScheduler(logger, rdb, 5*time.Second, queues)
-	subscriber := newSubscriber(logger, rdb, cancels)
-	processor := newProcessor(newProcessorParams{
+
+	syncer := newSyncer(syncerParams{
+		logger:     logger,
+		requestsCh: syncCh,
+		interval:   5 * time.Second,
+	})
+	heartbeater := newHeartbeater(heartbeaterParams{
+		logger:      logger,
+		broker:      rdb,
+		serverState: ss,
+		interval:    5 * time.Second,
+	})
+	scheduler := newScheduler(schedulerParams{
+		logger:   logger,
+		broker:   rdb,
+		interval: 5 * time.Second,
+		queues:   queues,
+	})
+	subscriber := newSubscriber(subscriberParams{
+		logger:       logger,
+		broker:       rdb,
+		cancelations: cancels,
+	})
+	processor := newProcessor(processorParams{
 		logger:          logger,
 		broker:          rdb,
 		ss:              ss,
@@ -291,13 +399,6 @@ func (srv *Server) Start(handler Handler) error {
 	srv.ss.SetStatus(base.StatusRunning)
 	srv.processor.handler = handler
 
-	type prefixLogger interface {
-		SetPrefix(prefix string)
-	}
-	// If logger supports setting prefix, then set prefix for log output.
-	if l, ok := srv.logger.(prefixLogger); ok {
-		l.SetPrefix(fmt.Sprintf("asynq: pid=%d ", os.Getpid()))
-	}
 	srv.logger.Info("Starting processing")
 
 	srv.heartbeater.start(&srv.wg)
@@ -319,7 +420,6 @@ func (srv *Server) Stop() {
 		return
 	}
 
-	fmt.Println() // print newline for prettier log.
 	srv.logger.Info("Starting graceful shutdown")
 	// Note: The order of termination is important.
 	// Sender goroutines should be terminated before the receiver goroutines.
@@ -335,12 +435,14 @@ func (srv *Server) Stop() {
 	srv.broker.Close()
 	srv.ss.SetStatus(base.StatusStopped)
 
-	srv.logger.Info("Bye!")
+	srv.logger.Info("Exiting")
 }
 
 // Quiet signals the server to stop pulling new tasks off queues.
 // Quiet should be used before stopping the server.
 func (srv *Server) Quiet() {
+	srv.logger.Info("Stopping processor")
 	srv.processor.stop()
 	srv.ss.SetStatus(base.StatusQuiet)
+	srv.logger.Info("Processor stopped")
 }

server_test.go

@@ -28,6 +28,7 @@ func TestServer(t *testing.T) {
 	c := NewClient(r)
 	srv := NewServer(r, Config{
 		Concurrency: 10,
+		LogLevel:    testLogLevel,
 	})
 
 	// no-op handler
@@ -58,7 +59,7 @@ func TestServerRun(t *testing.T) {
 	ignoreOpt := goleak.IgnoreTopFunction("github.com/go-redis/redis/v7/internal/pool.(*ConnPool).reaper")
 	defer goleak.VerifyNoLeaks(t, ignoreOpt)
 
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 
 	done := make(chan struct{})
 	// Make sure server exits when receiving TERM signal.
@@ -83,7 +84,7 @@ func TestServerRun(t *testing.T) {
 }
 
 func TestServerErrServerStopped(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	handler := NewServeMux()
 	if err := srv.Start(handler); err != nil {
 		t.Fatal(err)
@@ -96,7 +97,7 @@ func TestServerErrServerStopped(t *testing.T) {
 }
 
 func TestServerErrNilHandler(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	err := srv.Start(nil)
 	if err == nil {
 		t.Error("Starting server with nil handler: (*Server).Start(nil) did not return error")
@@ -105,7 +106,7 @@ func TestServerErrNilHandler(t *testing.T) {
 }
 
 func TestServerErrServerRunning(t *testing.T) {
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	handler := NewServeMux()
 	if err := srv.Start(handler); err != nil {
 		t.Fatal(err)
@@ -126,7 +127,7 @@ func TestServerWithRedisDown(t *testing.T) {
 	}()
 	r := rdb.NewRDB(setup(t))
 	testBroker := testbroker.NewTestBroker(r)
-	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: ":6379"}, Config{LogLevel: testLogLevel})
 	srv.broker = testBroker
 	srv.scheduler.broker = testBroker
 	srv.heartbeater.broker = testBroker
@@ -158,7 +159,7 @@ func TestServerWithFlakyBroker(t *testing.T) {
 	}()
 	r := rdb.NewRDB(setup(t))
 	testBroker := testbroker.NewTestBroker(r)
-	srv := NewServer(RedisClientOpt{Addr: redisAddr, DB: redisDB}, Config{})
+	srv := NewServer(RedisClientOpt{Addr: redisAddr, DB: redisDB}, Config{LogLevel: testLogLevel})
 	srv.broker = testBroker
 	srv.scheduler.broker = testBroker
 	srv.heartbeater.broker = testBroker
@@ -208,3 +209,32 @@ func TestServerWithFlakyBroker(t *testing.T) {
 
 	srv.Stop()
 }
+
+func TestLogLevel(t *testing.T) {
+	tests := []struct {
+		flagVal string
+		want    LogLevel
+		wantStr string
+	}{
+		{"debug", DebugLevel, "debug"},
+		{"Info", InfoLevel, "info"},
+		{"WARN", WarnLevel, "warn"},
+		{"warning", WarnLevel, "warn"},
+		{"Error", ErrorLevel, "error"},
+		{"fatal", FatalLevel, "fatal"},
+	}
+
+	for _, tc := range tests {
+		level := new(LogLevel)
+		if err := level.Set(tc.flagVal); err != nil {
+			t.Fatal(err)
+		}
+		if *level != tc.want {
+			t.Errorf("Set(%q): got %v, want %v", tc.flagVal, level, &tc.want)
+			continue
+		}
+		if got := level.String(); got != tc.wantStr {
+			t.Errorf("String() returned %q, want %q", got, tc.wantStr)
+		}
+	}
+}

subscriber.go

@@ -10,10 +10,11 @@ import (
 
 	"github.com/go-redis/redis/v7"
 	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/log"
 )
 
 type subscriber struct {
-	logger Logger
+	logger *log.Logger
 	broker base.Broker
 
 	// channel to communicate back to the long running "subscriber" goroutine.
@@ -26,18 +27,24 @@ type subscriber struct {
 	retryTimeout time.Duration
 }
 
-func newSubscriber(l Logger, b base.Broker, cancelations *base.Cancelations) *subscriber {
+type subscriberParams struct {
+	logger       *log.Logger
+	broker       base.Broker
+	cancelations *base.Cancelations
+}
+
+func newSubscriber(params subscriberParams) *subscriber {
 	return &subscriber{
-		logger:       l,
-		broker:       b,
+		logger:       params.logger,
+		broker:       params.broker,
 		done:         make(chan struct{}),
-		cancelations: cancelations,
+		cancelations: params.cancelations,
 		retryTimeout: 5 * time.Second,
 	}
 }
 
 func (s *subscriber) terminate() {
-	s.logger.Info("Subscriber shutting down...")
+	s.logger.Debug("Subscriber shutting down...")
 	// Signal the subscriber goroutine to stop.
 	s.done <- struct{}{}
 }
@@ -54,12 +61,12 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
 		for {
 			pubsub, err = s.broker.CancelationPubSub()
 			if err != nil {
-				s.logger.Error("cannot subscribe to cancelation channel: %v", err)
+				s.logger.Errorf("cannot subscribe to cancelation channel: %v", err)
 				select {
 				case <-time.After(s.retryTimeout):
 					continue
 				case <-s.done:
-					s.logger.Info("Subscriber done")
+					s.logger.Debug("Subscriber done")
 					return
 				}
 			}
@@ -70,7 +77,7 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
 			select {
 			case <-s.done:
 				pubsub.Close()
-				s.logger.Info("Subscriber done")
+				s.logger.Debug("Subscriber done")
 				return
 			case msg := <-cancelCh:
 				cancel, ok := s.cancelations.Get(msg.Payload)

subscriber_test.go

@@ -38,7 +38,11 @@ func TestSubscriber(t *testing.T) {
 		cancelations := base.NewCancelations()
 		cancelations.Add(tc.registeredID, fakeCancelFunc)
 
-		subscriber := newSubscriber(testLogger, rdbClient, cancelations)
+		subscriber := newSubscriber(subscriberParams{
+			logger:       testLogger,
+			broker:       rdbClient,
+			cancelations: cancelations,
+		})
 		var wg sync.WaitGroup
 		subscriber.start(&wg)
 		defer subscriber.terminate()
@@ -75,7 +79,11 @@ func TestSubscriberWithRedisDown(t *testing.T) {
 	testBroker := testbroker.NewTestBroker(r)
 
 	cancelations := base.NewCancelations()
-	subscriber := newSubscriber(testLogger, testBroker, cancelations)
+	subscriber := newSubscriber(subscriberParams{
+		logger:       testLogger,
+		broker:       testBroker,
+		cancelations: cancelations,
+	})
 	subscriber.retryTimeout = 1 * time.Second // set shorter retry timeout for testing purpose.
 
 	testBroker.Sleep() // simulate a situation where subscriber cannot connect to redis.

syncer.go

@@ -7,12 +7,14 @@ package asynq
 import (
 	"sync"
 	"time"
+
+	"github.com/hibiken/asynq/internal/log"
 )
 
 // syncer is responsible for queuing up failed requests to redis and retry
 // those requests to sync state between the background process and redis.
 type syncer struct {
-	logger Logger
+	logger *log.Logger
 
 	requestsCh <-chan *syncRequest
 
@@ -28,17 +30,23 @@ type syncRequest struct {
 	errMsg string       // error message
 }
 
-func newSyncer(l Logger, requestsCh <-chan *syncRequest, interval time.Duration) *syncer {
+type syncerParams struct {
+	logger     *log.Logger
+	requestsCh <-chan *syncRequest
+	interval   time.Duration
+}
+
+func newSyncer(params syncerParams) *syncer {
 	return &syncer{
-		logger:     l,
-		requestsCh: requestsCh,
+		logger:     params.logger,
+		requestsCh: params.requestsCh,
 		done:       make(chan struct{}),
-		interval:   interval,
+		interval:   params.interval,
 	}
 }
 
 func (s *syncer) terminate() {
-	s.logger.Info("Syncer shutting down...")
+	s.logger.Debug("Syncer shutting down...")
 	// Signal the syncer goroutine to stop.
 	s.done <- struct{}{}
 }
@@ -57,7 +65,7 @@ func (s *syncer) start(wg *sync.WaitGroup) {
 						s.logger.Error(req.errMsg)
 					}
 				}
-				s.logger.Info("Syncer done")
+				s.logger.Debug("Syncer done")
 				return
 			case req := <-s.requestsCh:
 				requests = append(requests, req)

syncer_test.go

@@ -27,7 +27,11 @@ func TestSyncer(t *testing.T) {
 
 	const interval = time.Second
 	syncRequestCh := make(chan *syncRequest)
-	syncer := newSyncer(testLogger, syncRequestCh, interval)
+	syncer := newSyncer(syncerParams{
+		logger:     testLogger,
+		requestsCh: syncRequestCh,
+		interval:   interval,
+	})
 	var wg sync.WaitGroup
 	syncer.start(&wg)
 	defer syncer.terminate()
@@ -52,7 +56,11 @@ func TestSyncer(t *testing.T) {
 func TestSyncerRetry(t *testing.T) {
 	const interval = time.Second
 	syncRequestCh := make(chan *syncRequest)
-	syncer := newSyncer(testLogger, syncRequestCh, interval)
+	syncer := newSyncer(syncerParams{
+		logger:     testLogger,
+		requestsCh: syncRequestCh,
+		interval:   interval,
+	})
 
 	var wg sync.WaitGroup
 	syncer.start(&wg)