Merge pull request #296 from moov-io/feat-sarama-logger

kafka: support a debug sarama logger

Author: adamdecaf

cmd/achgateway/main.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/moov-io/achgateway"
 	"github.com/moov-io/achgateway/internal"
+	"github.com/moov-io/achgateway/internal/kafka"
 	"github.com/moov-io/achgateway/internal/service"
 	"github.com/moov-io/base/log"
 )
@@ -33,6 +34,8 @@ func main() {
 		Logger: log.NewDefaultLogger().Set("app", log.String("achgateway")).Set("version", log.String(achgateway.Version)),
 	}
 
+	kafka.EnableSaramaDebugLogging(env.Logger)
+
 	env, err := internal.NewEnvironment(env)
 	if err != nil {
 		env.Logger.Fatal().LogErrorf("Error loading up environment: %v", err)

docs/config.md

@@ -174,6 +174,8 @@ ACHGateway:
       [ Endpoint: <string> | default = "" ]
 ```
 
+> TIP: Set `SARAMA_DEBUG_LOGGING=yes` to enable debug logging for the IBM/sarama (kafka) library.
+
 ### Sharding
 ```yaml
   Sharding:

internal/kafka/kafka_test.go

@@ -0,0 +1,30 @@
+package kafka_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/moov-io/achgateway/internal/kafka"
+	"github.com/moov-io/achgateway/internal/service"
+	"github.com/moov-io/base/log"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSaramaDebugLogging(t *testing.T) {
+	t.Setenv("SARAMA_DEBUG_LOGGING", "yes")
+
+	buf, logger := log.NewBufferLogger()
+	kafka.EnableSaramaDebugLogging(logger)
+
+	cfg := service.KafkaConfig{
+		Brokers: []string{"127.0.0.1:55555"},
+	}
+
+	topic, err := kafka.OpenTopic(log.NewNopLogger(), &cfg)
+	require.Nil(t, topic)
+	require.Error(t, err)
+
+	fmt.Printf("\n%s\n", buf.String())
+	require.Contains(t, buf.String(), `msg="ClientID is the default of 'sarama',`)
+}

internal/kafka/logger.go

@@ -0,0 +1,49 @@
+package kafka
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/moov-io/base/log"
+	"github.com/moov-io/base/strx"
+
+	"github.com/IBM/sarama"
+)
+
+// SaramaLogger implements the sarama.StdLogger interface with a moov-io/base logger
+//
+// All messages are logged at a debug level.
+type SaramaLogger struct {
+	logger log.Logger
+}
+
+func NewSaramaLogger(logger log.Logger) *SaramaLogger {
+	return &SaramaLogger{
+		logger: logger,
+	}
+}
+
+var _ sarama.StdLogger = (&SaramaLogger{})
+
+func (l *SaramaLogger) Print(v ...interface{}) {
+	l.logger.Debug().Log(fmt.Sprint(v...))
+}
+
+func (l *SaramaLogger) Printf(format string, v ...interface{}) {
+	l.logger.Debug().Logf(format, v...)
+
+}
+
+func (l *SaramaLogger) Println(v ...interface{}) {
+	l.Print(v...)
+}
+
+// EnableSaramaDebugLogging overrides the default sarama logger (which discards everything)
+// with the provided moov-io/base logger.
+//
+// All messages are logged at a debug level.
+func EnableSaramaDebugLogging(logger log.Logger) {
+	if strx.Yes(os.Getenv("SARAMA_DEBUG_LOGGING")) {
+		sarama.Logger = NewSaramaLogger(logger)
+	}
+}