Logging refactoring (#771)

<!--
Thanks for taking precious time for making a PR.

Before creating a pull request, please make sure:
- Your PR solves one problem for which an issue exist and a solution has
been discussed
- You have read the guide for contributing
- See
https://github.com/beatlabs/patron/blob/master/README.md#how-to-contribute
- You signed all your commits (otherwise we won't be able to merge the
PR)
  - See https://github.com/beatlabs/patron/blob/master/SIGNYOURWORK.md
- You added unit tests for the new functionality
- You mention in the PR description which issue it is addressing, e.g.
"Resolves #123"
-->

## Which problem is this PR solving?

Resolves #155.

## Short description of the changes

<!-- REQUIRED -->

component/amqp/message_test.go

@@ -27,7 +27,9 @@ var (
 )
 
 func TestMain(m *testing.M) {
-	os.Setenv("OTEL_BSP_SCHEDULE_DELAY", "100")
+	if err := os.Setenv("OTEL_BSP_SCHEDULE_DELAY", "100"); err != nil {
+		panic(err)
+	}
 
 	tracePublisher = patrontrace.Setup("test", nil, traceExporter)
 

component/http/observability.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/pprof"
+
+	"github.com/beatlabs/patron/observability/log"
 )
 
 func ProfilingRoutes(enableExpVar bool) []*Route {
@@ -47,3 +49,24 @@ func expVars(w http.ResponseWriter, _ *http.Request) {
 	})
 	_, _ = fmt.Fprintf(w, "\n}\n")
 }
+
+// LoggingRoutes returns a routes relates to logs.
+func LoggingRoutes() []*Route {
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		lvl := r.PathValue("level")
+		if lvl == "" {
+			http.Error(w, "missing log level", http.StatusBadRequest)
+			return
+		}
+
+		err := log.SetLevel(lvl)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+	}
+
+	route, _ := NewRoute("POST /debug/log/{level}", handler)
+	return []*Route{route}
+}

component/http/observability_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/beatlabs/patron/observability/log"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -18,7 +19,7 @@ type profilingTestCase struct {
 
 func TestProfilingRoutes(t *testing.T) {
 	t.Run("without vars", func(t *testing.T) {
-		server := createProfilingServer(false)
+		server := createServer(false)
 		defer server.Close()
 
 		for name, tt := range createProfilingTestCases(false) {
@@ -34,7 +35,7 @@ func TestProfilingRoutes(t *testing.T) {
 	})
 
 	t.Run("with vars", func(t *testing.T) {
-		server := createProfilingServer(true)
+		server := createServer(true)
 		defer server.Close()
 
 		for name, tt := range createProfilingTestCases(true) {
@@ -50,11 +51,14 @@ func TestProfilingRoutes(t *testing.T) {
 	})
 }
 
-func createProfilingServer(enableExpVar bool) *httptest.Server {
+func createServer(enableExpVar bool) *httptest.Server {
 	mux := http.NewServeMux()
 	for _, route := range ProfilingRoutes(enableExpVar) {
 		mux.HandleFunc(route.path, route.handler)
 	}
+	for _, route := range LoggingRoutes() {
+		mux.HandleFunc(route.path, route.handler)
+	}
 
 	return httptest.NewServer(mux)
 }
@@ -80,3 +84,39 @@ func createProfilingTestCases(enableExpVar bool) map[string]profilingTestCase {
 		"vars":         {"/debug/vars/", expVarWant},
 	}
 }
+
+func TestLoggingRoutes(t *testing.T) {
+	require.NoError(t, log.Setup(&log.Config{
+		IsJSON: true,
+		Level:  "info",
+	}))
+	server := createServer(true)
+	defer server.Close()
+
+	t.Run("change log level to debug", func(t *testing.T) {
+		req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, server.URL+"/debug/log/debug", nil)
+		require.NoError(t, err)
+		resp, err := http.DefaultClient.Do(req)
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+		require.NoError(t, resp.Body.Close())
+	})
+
+	t.Run("wrong log level", func(t *testing.T) {
+		req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, server.URL+"/debug/log/xxx", nil)
+		require.NoError(t, err)
+		resp, err := http.DefaultClient.Do(req)
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+		require.NoError(t, resp.Body.Close())
+	})
+
+	t.Run("empty log level", func(t *testing.T) {
+		req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, server.URL+"/debug/log/", nil)
+		require.NoError(t, err)
+		resp, err := http.DefaultClient.Do(req)
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusNotFound, resp.StatusCode)
+		require.NoError(t, resp.Body.Close())
+	})
+}

component/kafka/component_test.go

@@ -26,7 +26,9 @@ var (
 )
 
 func TestMain(m *testing.M) {
-	os.Setenv("OTEL_BSP_SCHEDULE_DELAY", "100")
+	if err := os.Setenv("OTEL_BSP_SCHEDULE_DELAY", "100"); err != nil {
+		panic(err)
+	}
 
 	tracePublisher = patrontrace.Setup("test", nil, traceExporter)
 	os.Exit(m.Run())
@@ -39,7 +41,7 @@ func TestNew(t *testing.T) {
 	// consumer will commit every batch in a blocking operation
 	saramaCfg.Consumer.Offsets.AutoCommit.Enable = false
 	saramaCfg.Consumer.Offsets.Initial = sarama.OffsetOldest
-	saramaCfg.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
+	saramaCfg.Consumer.Group.Rebalance.GroupStrategies = append(saramaCfg.Consumer.Group.Rebalance.GroupStrategies, sarama.NewBalanceStrategySticky())
 	saramaCfg.Net.DialTimeout = 15 * time.Second
 	saramaCfg.Version = sarama.V2_6_0_0
 

component/sqs/component_test.go

@@ -22,7 +22,9 @@ var (
 )
 
 func TestMain(m *testing.M) {
-	os.Setenv("OTEL_BSP_SCHEDULE_DELAY", "100")
+	if err := os.Setenv("OTEL_BSP_SCHEDULE_DELAY", "100"); err != nil {
+		panic(err)
+	}
 
 	tracePublisher = patrontrace.Setup("test", nil, traceExporter)
 

component/sqs/metric.go

@@ -30,8 +30,8 @@ func init() {
 	messageQueueSizeGauge = patronmetric.Float64Gauge(packageName, "sqs.queue.size", "SQS message queue size.", "1")
 }
 
-func observerMessageAge(ctx context.Context, queue string, attributes map[string]string) {
-	attribute, ok := attributes[sqsAttributeSentTimestamp]
+func observerMessageAge(ctx context.Context, queue string, attrs map[string]string) {
+	attribute, ok := attrs[sqsAttributeSentTimestamp]
 	if !ok || len(strings.TrimSpace(attribute)) == 0 {
 		return
 	}

examples/service/http.go

@@ -32,10 +32,10 @@ func createHttpRouter() (patron.Component, error) {
 		return nil, fmt.Errorf("failed to create routes: %w", err)
 	}
 
-	router, err := router.New(router.WithRoutes(rr...))
+	rt, err := router.New(router.WithRoutes(rr...))
 	if err != nil {
 		return nil, fmt.Errorf("failed to create http router: %w", err)
 	}
 
-	return patronhttp.New(router)
+	return patronhttp.New(rt)
 }

observability/integration_test.go

@@ -6,14 +6,19 @@ import (
 	"context"
 	"testing"
 
+	"github.com/beatlabs/patron/observability/log"
 	"github.com/stretchr/testify/require"
 )
 
 func TestSetup(t *testing.T) {
 	t.Setenv("OTEL_EXPORTER_OTLP_INSECURE", "true")
 	ctx := context.Background()
 
-	got, err := Setup(ctx, "test", "1.2.3")
+	got, err := Setup(ctx, Config{
+		LogConfig: log.Config{
+			Level: "debug",
+		},
+	})
 	require.NoError(t, err)
 
 	require.NoError(t, got.Shutdown(ctx))

observability/log/log.go

@@ -4,10 +4,64 @@ package log
 import (
 	"context"
 	"log/slog"
+	"os"
 )
 
+// Config represents the configuration for setting up the logger.
+type Config struct {
+	Attributes []slog.Attr
+	IsJSON     bool
+	Level      string
+}
+
 type ctxKey struct{}
 
+var logCfg *Config
+
+// Setup sets up the logger with the given configuration.
+func Setup(cfg *Config) error {
+	logCfg = cfg
+	return setDefaultLogger(cfg)
+}
+
+// SetLevel sets the logger level.
+func SetLevel(lvl string) error {
+	logCfg.Level = lvl
+	return setDefaultLogger(logCfg)
+}
+
+func setDefaultLogger(cfg *Config) error {
+	lvl, err := level(cfg.Level)
+	if err != nil {
+		return err
+	}
+
+	ho := &slog.HandlerOptions{
+		AddSource: true,
+		Level:     lvl,
+	}
+
+	var hnd slog.Handler
+
+	if cfg.IsJSON {
+		hnd = slog.NewJSONHandler(os.Stderr, ho)
+	} else {
+		hnd = slog.NewTextHandler(os.Stderr, ho)
+	}
+
+	slog.SetDefault(slog.New(hnd.WithAttrs(cfg.Attributes)))
+	return nil
+}
+
+func level(lvl string) (slog.Level, error) {
+	lv := slog.LevelVar{}
+	if err := lv.UnmarshalText([]byte(lvl)); err != nil {
+		return slog.LevelInfo, err
+	}
+
+	return lv.Level(), nil
+}
+
 // FromContext returns the logger, if it exists in the context, or nil.
 func FromContext(ctx context.Context) *slog.Logger {
 	if l, ok := ctx.Value(ctxKey{}).(*slog.Logger); ok {

observability/log/log_test.go

@@ -7,8 +7,31 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
+func TestSetup(t *testing.T) {
+	t.Run("JSON", func(t *testing.T) {
+		cfg := &Config{
+			Attributes: []slog.Attr{},
+			IsJSON:     true,
+			Level:      "debug",
+		}
+		require.NoError(t, Setup(cfg))
+		assert.NotNil(t, slog.Default())
+	})
+
+	t.Run("Text", func(t *testing.T) {
+		cfg := &Config{
+			Attributes: []slog.Attr{},
+			IsJSON:     false,
+			Level:      "debug",
+		}
+		require.NoError(t, Setup(cfg))
+		assert.NotNil(t, slog.Default())
+	})
+}
+
 func TestContext(t *testing.T) {
 	l := slog.Default()
 
@@ -23,22 +46,19 @@ func TestContext(t *testing.T) {
 	})
 }
 
-func TestEnabled(t *testing.T) {
-	type args struct {
-		l slog.Level
-	}
-	tests := map[string]struct {
-		args args
-		want bool
-	}{
-		"Disabled": {args{slog.LevelDebug}, false},
-		"Enabled":  {args{slog.LevelInfo}, true},
-	}
-	for name, tt := range tests {
-		t.Run(name, func(t *testing.T) {
-			assert.Equal(t, tt.want, Enabled(tt.args.l))
-		})
-	}
+func TestSetLevelAndCheckEnable(t *testing.T) {
+	require.NoError(t, Setup(&Config{
+		Attributes: []slog.Attr{},
+		IsJSON:     true,
+		Level:      "info",
+	}))
+
+	assert.True(t, Enabled(slog.LevelInfo))
+	assert.False(t, Enabled(slog.LevelDebug))
+
+	require.NoError(t, SetLevel("debug"))
+
+	assert.True(t, Enabled(slog.LevelDebug))
 }
 
 func TestErrorAttr(t *testing.T) {

observability/observability.go

@@ -49,11 +49,23 @@ func StatusAttribute(err error) attribute.KeyValue {
 	return SucceededAttribute
 }
 
+// Config represents the configuration for setting up traces, metrics and logs.
+type Config struct {
+	Name      string
+	Version   string
+	LogConfig log.Config
+}
+
 // Setup initializes OpenTelemetry's traces and metrics.
 // It creates a resource with the given name and version, sets up the metric and trace providers,
 // and returns a Provider containing the initialized providers.
-func Setup(ctx context.Context, name, version string) (*Provider, error) {
-	res, err := createResource(name, version)
+func Setup(ctx context.Context, cfg Config) (*Provider, error) {
+	err := log.Setup(&cfg.LogConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := createResource(cfg.Name, cfg.Version)
 	if err != nil {
 		return nil, err
 	}
@@ -64,7 +76,7 @@ func Setup(ctx context.Context, name, version string) (*Provider, error) {
 	if err != nil {
 		return nil, err
 	}
-	traceProvider, err := patrontrace.SetupGRPC(ctx, name, res)
+	traceProvider, err := patrontrace.SetupGRPC(ctx, cfg.Name, res)
 	if err != nil {
 		return nil, err
 	}

options.go

@@ -34,7 +34,7 @@ func WithLogFields(attrs ...slog.Attr) OptionFunc {
 				continue
 			}
 
-			svc.logConfig.attrs = append(svc.logConfig.attrs, attr)
+			svc.observabilityCfg.LogConfig.Attributes = append(svc.observabilityCfg.LogConfig.Attributes, attr)
 		}
 
 		return nil
@@ -44,7 +44,7 @@ func WithLogFields(attrs ...slog.Attr) OptionFunc {
 // WithJSONLogger to use Go's slog package.
 func WithJSONLogger() OptionFunc {
 	return func(svc *Service) error {
-		svc.logConfig.json = true
+		svc.observabilityCfg.LogConfig.IsJSON = true
 		return nil
 	}
 }