Make logger configurable

pkg/conduit/config.go

@@ -70,8 +70,9 @@ type Config struct {
 	}
 
 	Log struct {
-		Level  string
-		Format string
+		NewLogger func(level, format string) log.CtxLogger
+		Level     string
+		Format    string
 	}
 
 	Connectors struct {
@@ -124,6 +125,7 @@ func DefaultConfig() Config {
 	cfg.API.HTTP.Address = ":8080"
 	cfg.API.GRPC.Address = ":8084"
 
+	cfg.Log.NewLogger = newLogger
 	cfg.Log.Level = "info"
 	cfg.Log.Format = "cli"
 

pkg/conduit/runtime.go

@@ -111,7 +111,12 @@ func NewRuntime(cfg Config) (*Runtime, error) {
 		return nil, cerrors.Errorf("invalid config: %w", err)
 	}
 
-	logger := newLogger(cfg.Log.Level, cfg.Log.Format)
+	logger := cfg.Log.NewLogger(cfg.Log.Level, cfg.Log.Format)
+	logger.Logger = logger.
+		Hook(ctxutil.MessageIDLogCtxHook{}).
+		Hook(ctxutil.RequestIDLogCtxHook{}).
+		Hook(ctxutil.FilepathLogCtxHook{})
+	zerolog.DefaultContextLogger = &logger.Logger
 
 	var db database.DB
 	db = cfg.DB.Driver
@@ -254,13 +259,7 @@ func newLogger(level string, format string) log.CtxLogger {
 	// TODO make logger hooks configurable
 	l, _ := zerolog.ParseLevel(level)
 	f, _ := log.ParseFormat(format)
-	logger := log.InitLogger(l, f)
-	logger.Logger = logger.
-		Hook(ctxutil.MessageIDLogCtxHook{}).
-		Hook(ctxutil.RequestIDLogCtxHook{}).
-		Hook(ctxutil.FilepathLogCtxHook{})
-	zerolog.DefaultContextLogger = &logger.Logger
-	return logger
+	return log.InitLogger(l, f)
 }
 
 func configurePrometheus() {

pkg/conduit/runtime_test.go

@@ -15,23 +15,47 @@
 package conduit_test
 
 import (
+	"bytes"
 	"context"
+	"io"
+	"os"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/conduitio/conduit-commons/cchan"
 	"github.com/conduitio/conduit/pkg/conduit"
 	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/log"
+
 	"github.com/matryer/is"
+	"github.com/rs/zerolog"
 )
 
 func TestRuntime(t *testing.T) {
-	is := is.New(t)
+	var (
+		is   = is.New(t)
+		logs = safeBuffer{}
+		w    = io.MultiWriter(&logs, os.Stdout)
+	)
 
 	cfg := conduit.DefaultConfig()
 	cfg.DB.Badger.Path = t.TempDir() + "/testing.app.db"
 	cfg.API.GRPC.Address = ":0"
 	cfg.API.HTTP.Address = ":0"
+	cfg.Log.NewLogger = func(level, _ string) log.CtxLogger {
+		l, _ := zerolog.ParseLevel(level)
+		zl := zerolog.New(w).
+			With().
+			Timestamp().
+			Stack().
+			Logger().
+			Level(l)
+
+		return log.New(zl)
+	}
+
 	r, err := conduit.NewRuntime(cfg)
 	is.NoErr(err)
 	is.True(r != nil)
@@ -53,4 +77,35 @@ func TestRuntime(t *testing.T) {
 	if !cerrors.Is(err, context.Canceled) {
 		t.Logf("expected error '%v', got '%v'", context.Canceled, err)
 	}
+	is.True(strings.Contains(logs.String(), "grpc API started"))
+}
+
+// safeBuffer wraps bytes.Buffer and makes it safe for concurrent use.
+type safeBuffer struct {
+	b bytes.Buffer
+	m sync.RWMutex
+}
+
+func (b *safeBuffer) Read(p []byte) (n int, err error) {
+	b.m.Lock()
+	defer b.m.Unlock()
+	return b.b.Read(p)
+}
+
+func (b *safeBuffer) Write(p []byte) (n int, err error) {
+	b.m.Lock()
+	defer b.m.Unlock()
+	return b.b.Write(p)
+}
+
+func (b *safeBuffer) String() string {
+	b.m.RLock()
+	defer b.m.RUnlock()
+	return b.b.String()
+}
+
+func (b *safeBuffer) Len() int {
+	b.m.RLock()
+	defer b.m.RUnlock()
+	return b.b.Len()
 }