[feat] added a slog wrapper of the logger interface (#1234)

### Motivation

This commit supports to use `log/slog` package from the standard library to control the level and output type of the logs.
In order for us to not have to import logrus as a direct dependency for part of our testing suit, it would be nice if we can use `slog` package instead, and wrap that in the provided by `pulsar/log` interfaces.

This ties in a bit with issue #1078 because it opens the door for users who are already working with log/slog in their projects. Plus, it's a gives more time for the Pulsar team to evaluate incorporating slog into the SDK.

### Modifications

One additional file `/pulsar/log/wrapper_slog.go` is added.
One additional function in the `pulsar/log` package, `NewLoggerWithSlog` , is exposed.

README.md

@@ -152,7 +152,7 @@ Run the tests:
 
 Run the tests with specific versions of GOLANG and PULSAR:
 
-    make test GOLANG_VERSION=1.20 PULSAR_VERSION=2.10.0
+    make test GO_VERSION=1.20 PULSAR_VERSION=2.10.0
 
 ## Contributing
 

go.mod

@@ -1,6 +1,6 @@
 module github.com/apache/pulsar-client-go
 
-go 1.18
+go 1.20
 
 require (
 	github.com/99designs/keyring v1.2.1

pulsar/client_impl_with_slog_test.go

@@ -0,0 +1,49 @@
+/// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build go1.21
+
+package pulsar
+
+import (
+	"log/slog"
+	"os"
+	"testing"
+
+	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClientWithSlog(t *testing.T) {
+	sLogger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+
+	client, err := NewClient(ClientOptions{
+		URL:    serviceURL,
+		Logger: log.NewLoggerWithSlog(sLogger),
+	})
+	assert.NotNil(t, client)
+	assert.Nil(t, err)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newTopicName(),
+	})
+	assert.NotNil(t, producer)
+	assert.Nil(t, err)
+
+	producer.Close()
+	client.Close()
+}

pulsar/log/wrapper_slog.go

@@ -0,0 +1,105 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build go1.21
+
+package log
+
+import (
+	"fmt"
+	"log/slog"
+)
+
+type slogWrapper struct {
+	logger *slog.Logger
+}
+
+func (s *slogWrapper) Debug(args ...any) {
+	message := s.tryDetermineMessage(args...)
+	s.logger.Debug(message)
+}
+
+func (s *slogWrapper) Info(args ...any) {
+	message := s.tryDetermineMessage(args...)
+	s.logger.Info(message)
+}
+
+func (s *slogWrapper) Error(args ...any) {
+	message := s.tryDetermineMessage(args...)
+	s.logger.Error(message)
+}
+
+func (s *slogWrapper) Warn(args ...any) {
+	message := s.tryDetermineMessage(args...)
+	s.logger.Warn(message)
+}
+
+func (s *slogWrapper) Debugf(format string, args ...any) {
+	s.logger.Debug(fmt.Sprintf(format, args...))
+}
+
+func (s *slogWrapper) Infof(format string, args ...any) {
+	s.logger.Info(fmt.Sprintf(format, args...))
+}
+
+func (s *slogWrapper) Warnf(format string, args ...any) {
+	s.logger.Warn(fmt.Sprintf(format, args...))
+}
+
+func (s *slogWrapper) Errorf(format string, args ...any) {
+	s.logger.Error(fmt.Sprintf(format, args...))
+}
+
+func (s *slogWrapper) SubLogger(fields Fields) Logger {
+	return &slogWrapper{
+		logger: s.logger.With(pulsarFieldsToKVSlice(fields)...),
+	}
+}
+
+func (s *slogWrapper) WithError(err error) Entry {
+	return s.WithField("error", err)
+}
+
+func (s *slogWrapper) WithField(name string, value any) Entry {
+	return &slogWrapper{
+		logger: s.logger.With(name, value),
+	}
+}
+
+func (s *slogWrapper) WithFields(fields Fields) Entry {
+	return &slogWrapper{
+		logger: s.logger.With(pulsarFieldsToKVSlice(fields)...),
+	}
+}
+
+func NewLoggerWithSlog(logger *slog.Logger) Logger {
+	return &slogWrapper{
+		logger: logger,
+	}
+}
+
+func pulsarFieldsToKVSlice(f Fields) []any {
+	ret := make([]any, 0, len(f)*2)
+	for k, v := range f {
+		ret = append(ret, k, v)
+	}
+	return ret
+}
+
+func (s *slogWrapper) tryDetermineMessage(value ...any) string {
+	return fmt.Sprint(value...)
+}

pulsar/log/wrapper_slog_test.go

@@ -0,0 +1,186 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build go1.21
+
+package log
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSlogLevels(t *testing.T) {
+	testCases := []struct {
+		level       slog.Level
+		logFunction func(logger Logger, msg string)
+	}{
+		{slog.LevelDebug, func(logger Logger, msg string) { logger.Debug(msg) }},
+		{slog.LevelInfo, func(logger Logger, msg string) { logger.Info(msg) }},
+		{slog.LevelWarn, func(logger Logger, msg string) { logger.Warn(msg) }},
+		{slog.LevelError, func(logger Logger, msg string) { logger.Error(msg) }},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.level.String(), func(t *testing.T) {
+			var logBuffer bytes.Buffer
+			logMessage := "test message"
+			loggerSlog := slog.New(slog.NewJSONHandler(&logBuffer, &slog.HandlerOptions{Level: tc.level}))
+			pulsarLogger := NewLoggerWithSlog(loggerSlog)
+
+			tc.logFunction(pulsarLogger, logMessage)
+
+			logOutputSlog := logBuffer.String()
+			verifyLogOutput(t, logOutputSlog, tc.level.String(), logMessage)
+		})
+	}
+}
+
+func TestSlogPrintMethods(t *testing.T) {
+	testCases := []struct {
+		level       slog.Level
+		logFunction func(logger Logger, format string, args ...any)
+	}{
+		{
+			level: slog.LevelDebug,
+			logFunction: func(logger Logger, format string, args ...any) {
+				logger.Debugf(format, args...)
+			},
+		},
+		{
+			level: slog.LevelInfo,
+			logFunction: func(logger Logger, format string, args ...any) {
+				logger.Infof(format, args...)
+			},
+		},
+		{
+			level: slog.LevelWarn,
+			logFunction: func(logger Logger, format string, args ...any) {
+				logger.Warnf(format, args...)
+			},
+		},
+		{
+			level: slog.LevelError,
+			logFunction: func(logger Logger, format string, args ...any) {
+				logger.Errorf(format, args...)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.level.String()+"f", func(t *testing.T) {
+			var logBuffer bytes.Buffer
+			logMessage := "formatted message for %s"
+			expectedMessage := "formatted message for " + tc.level.String()
+			loggerSlog := slog.New(slog.NewJSONHandler(&logBuffer, &slog.HandlerOptions{Level: tc.level}))
+			pulsarLogger := NewLoggerWithSlog(loggerSlog)
+
+			tc.logFunction(pulsarLogger, logMessage, tc.level.String())
+
+			logOutputSlog := logBuffer.String()
+			verifyLogOutput(t, logOutputSlog, tc.level.String(), expectedMessage)
+		})
+	}
+}
+
+func TestSlogWrapperWithMethods(t *testing.T) {
+	testCases := []struct {
+		name           string
+		level          slog.Level
+		testMessage    string
+		setupLogger    func(logger Logger) Entry
+		expectedFields Fields
+	}{
+		{
+			name:        "WithField",
+			level:       slog.LevelInfo,
+			testMessage: "Message with field",
+			setupLogger: func(logger Logger) Entry {
+				return logger.WithField("key", "value")
+			},
+			expectedFields: Fields{"key": "value"},
+		},
+		{
+			name:        "WithFields",
+			level:       slog.LevelInfo,
+			testMessage: "Message with multiple fields",
+			setupLogger: func(logger Logger) Entry {
+				return logger.WithFields(Fields{"key1": "value1", "key2": "value2"})
+			},
+			expectedFields: Fields{"key1": "value1", "key2": "value2"},
+		},
+		{
+			name:        "WithError",
+			level:       slog.LevelInfo,
+			testMessage: "Message with error field",
+			setupLogger: func(logger Logger) Entry {
+				return logger.WithError(errors.New("test error"))
+			},
+			expectedFields: Fields{"error": "test error"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var logBuffer bytes.Buffer
+			loggerSlog := slog.New(slog.NewJSONHandler(&logBuffer, &slog.HandlerOptions{Level: tc.level}))
+			pulsarLogger := NewLoggerWithSlog(loggerSlog)
+
+			entry := tc.setupLogger(pulsarLogger)
+			switch tc.level {
+			case slog.LevelDebug:
+				entry.Debug(tc.testMessage)
+			case slog.LevelInfo:
+				entry.Info(tc.testMessage)
+			case slog.LevelWarn:
+				entry.Warn(tc.testMessage)
+			case slog.LevelError:
+				entry.Error(tc.testMessage)
+			default:
+				t.Errorf("Unsupported log level: %v", tc.level)
+			}
+
+			verifyLogOutput(t, logBuffer.String(), tc.level.String(), tc.testMessage, tc.expectedFields)
+		})
+	}
+}
+
+func verifyLogOutput(t *testing.T, logOutput, expectedLevel, expectedMessage string, expectedFields ...Fields) {
+	logLines := strings.Split(strings.TrimSpace(logOutput), "\n")
+	require.Len(t, logLines, 1, "There should be exactly one log line.")
+
+	var logEntry map[string]interface{}
+	err := json.Unmarshal([]byte(logLines[0]), &logEntry)
+	require.NoError(t, err, "Log entry should be valid JSON.")
+	require.Equal(t, expectedLevel, logEntry[slog.LevelKey], "Log level should match expected level.")
+	require.Equal(t, expectedMessage, logEntry[slog.MessageKey], "Log message should contain expected message.")
+
+	if len(expectedFields) > 0 {
+		for key, expectedValue := range expectedFields[0] {
+			actualValue, ok := logEntry[key]
+			require.True(t, ok, fmt.Sprintf("Expected key '%s' to be present in the log entry", key))
+			require.Equal(t, expectedValue, actualValue, fmt.Sprintf("Value for key '%s' should match the expected value", key))
+		}
+	}
+}