Fix closerCore With behaviour

This commit fixes the behaviour of `closerCore.With` and add tests to
ensure it works as expected.

The tests are a modification of existing tests that calls the
configuration functions like Beats does and ensures both closerCore
and typedCore are used. The method calls get proxyed from one core to
another, so both are tested together.

logp/core.go

@@ -26,7 +26,6 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"unsafe"
 
@@ -67,14 +66,13 @@ type coreLogger struct {
 type closerCore struct {
 	zapcore.Core
 	io.Closer
-	mutex sync.Mutex
 }
 
-func (c *closerCore) With(fields []zapcore.Field) zapcore.Core {
-	c.mutex.Lock()
-	c.Core = c.Core.With(fields)
-	c.mutex.Unlock()
-	return c
+func (c closerCore) With(fields []zapcore.Field) zapcore.Core {
+	return closerCore{
+		Core:   c.Core.With(fields),
+		Closer: c.Closer,
+	}
 }
 
 // Configure configures the logp package.

logp/core_test.go

@@ -635,6 +635,70 @@ func TestCoresCanBeClosed(t *testing.T) {
 	}
 }
 
+func TestCloserLoggerCoreWith(t *testing.T) {
+	defaultWriter := writeSyncer{}
+
+	cfg := zap.NewProductionEncoderConfig()
+	cfg.TimeKey = "" // remove the time to make the log entry consistent
+
+	core := closerCore{
+		Core: zapcore.NewCore(
+			zapcore.NewJSONEncoder(cfg),
+			&defaultWriter,
+			zapcore.InfoLevel,
+		),
+	}
+
+	expectedLines := []string{
+		// First/Default logger
+		`{"level":"info","msg":"Very first message"}`,
+
+		// Two messages after calling With
+		`{"level":"info","msg":"a message with extra fields","foo":"bar"}`,
+		`{"level":"info","msg":"another message with extra fields","foo":"bar"}`,
+
+		// A message with the default logger
+		`{"level":"info","msg":"a message without extra fields"}`,
+
+		// Two more messages with a different field
+		`{"level":"info","msg":"a message with an answer","answer":"42"}`,
+		`{"level":"info","msg":"another message with an answer","answer":"42"}`,
+
+		// One last message with the default logger
+		`{"level":"info","msg":"another message without any extra fields"}`,
+	}
+
+	// The default logger, it should not be modified by any call to With.
+	logger := zap.New(&core)
+	logger.Info("Very first message")
+
+	// Add a field and write messages
+	loggerWithFields := logger.With(strField("foo", "bar"))
+	loggerWithFields.Info("a message with extra fields")
+	loggerWithFields.Info("another message with extra fields")
+
+	// Use the default logger again
+	logger.Info("a message without extra fields")
+
+	// New logger with other fields
+	loggerWithFields = logger.With(strField("answer", "42"))
+	loggerWithFields.Info("a message with an answer")
+	loggerWithFields.Info("another message with an answer")
+
+	// One last message with the default logger
+	logger.Info("another message without any extra fields")
+
+	scanner := bufio.NewScanner(strings.NewReader(defaultWriter.String()))
+	count := 0
+	for scanner.Scan() {
+		l := scanner.Text()
+		if l != expectedLines[count] {
+			t.Error("Expecting:\n", l, "\nGot:\n", expectedLines[count])
+		}
+		count++
+	}
+}
+
 func strField(key, val string) zapcore.Field {
 	return zapcore.Field{Type: zapcore.StringType, Key: key, String: val}
 }

logp/selective_test.go

@@ -18,6 +18,11 @@
 package logp
 
 import (
+	"bufio"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -55,24 +60,28 @@ func TestLoggerSelectors(t *testing.T) {
 	assert.Len(t, logs, 1)
 }
 
-func TestTypedCoreSelectors(t *testing.T) {
+func TestTypedAndCloserCoreSelectors(t *testing.T) {
+	tempDir := t.TempDir()
+
 	logSelector := "enabled-log-selector"
 	expectedMsg := "this should be logged"
 
 	defaultCfg := DefaultConfig(DefaultEnvironment)
 	eventsCfg := DefaultEventConfig(DefaultEnvironment)
 
 	defaultCfg.Level = DebugLevel
-	defaultCfg.toObserver = true
-	defaultCfg.ToStderr = false
-	defaultCfg.ToFiles = false
+	defaultCfg.Beat = t.Name()
 	defaultCfg.Selectors = []string{logSelector}
+	defaultCfg.ToStderr = false
+	defaultCfg.ToFiles = true
+	defaultCfg.Files.Path = tempDir
 
 	eventsCfg.Level = defaultCfg.Level
-	eventsCfg.toObserver = defaultCfg.toObserver
+	eventsCfg.Beat = defaultCfg.Beat
+	eventsCfg.Selectors = defaultCfg.Selectors
 	eventsCfg.ToStderr = defaultCfg.ToStderr
 	eventsCfg.ToFiles = defaultCfg.ToFiles
-	eventsCfg.Selectors = defaultCfg.Selectors
+	eventsCfg.Files.Path = tempDir
 
 	if err := ConfigureWithTypedOutput(defaultCfg, eventsCfg, "log.type", "event"); err != nil {
 		t.Fatalf("could not configure logger: %s", err)
@@ -85,37 +94,65 @@ func TestTypedCoreSelectors(t *testing.T) {
 	enabledSelector.Debugw(expectedMsg, "log.type", "event")
 	disabledSelector.Debug("this should not be logged")
 
-	logEntries := ObserverLogs().TakeAll()
+	logEntries := takeAllLogsFromPath(t, tempDir)
 	if len(logEntries) != 2 {
 		t.Errorf("expecting 2 log entries, got %d", len(logEntries))
 		t.Log("Log entries:")
 		for _, e := range logEntries {
-			t.Log("Message:", e.Message, "Fields:", e.Context)
+			t.Log(e)
 		}
 		t.FailNow()
 	}
 
 	for i, logEntry := range logEntries {
-		msg := logEntry.Message
+		msg := logEntry["message"].(string)
 		if msg != expectedMsg {
 			t.Fatalf("[%d] expecting log message '%s', got '%s'", i, expectedMsg, msg)
 		}
 
 		// The second entry should also contain `log.type: event`
 		if i == 1 {
-			fields := logEntry.Context
-			if len(fields) != 1 {
-				t.Errorf("expecting one field, got %d", len(fields))
+			logType := logEntry["log.type"].(string)
+			if logType != "event" {
+				t.Errorf("expecting value 'event', got '%s'", logType)
 			}
+		}
+	}
+}
 
-			k := fields[0].Key
-			v := fields[0].String
-			if k != "log.type" {
-				t.Errorf("expecting key 'log.type', got '%s'", k)
-			}
-			if v != "event" {
-				t.Errorf("expecting value 'event', got '%s'", v)
+func takeAllLogsFromPath(t *testing.T, path string) []map[string]any {
+	entries := []map[string]any{}
+
+	glob := filepath.Join(path, "*.ndjson")
+	files, err := filepath.Glob(glob)
+	if err != nil {
+		t.Fatalf("cannot get files for glob '%s': %s", glob, err)
+	}
+
+	for _, filePath := range files {
+		f, err := os.Open(filePath)
+		if err != nil {
+			t.Fatalf("cannot open file '%s': %s", filePath, err)
+		}
+		defer f.Close()
+
+		sc := bufio.NewScanner(f)
+		for sc.Scan() {
+			m := map[string]any{}
+			bytes := sc.Bytes()
+			if err := json.Unmarshal(bytes, &m); err != nil {
+				t.Fatalf("cannot unmarshal log entry: '%s', err: %s", string(bytes), err)
 			}
+
+			entries = append(entries, m)
 		}
 	}
+
+	sort.Slice(entries, func(i, j int) bool {
+		t1 := entries[i]["@timestamp"].(string)
+		t2 := entries[j]["@timestamp"].(string)
+		return t1 < t2
+	})
+
+	return entries
 }