Merge #150558

150558: util/log: fix data race in (*StructuredLogSpy).Intercept  r=yuzefovich a=stevendanna

We were reading the formatter and filters without taking the lock. The
formatter doesn't appear to be set after initialization so I moved it
out of the lock struct. For the filters, I moved the locking up a few
lines.

Fixes #150490
Fixes #150556

Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/jobs/jobs_test.go

@@ -8,7 +8,6 @@ package jobs_test
 import (
 	"context"
 	gosql "database/sql"
-	"encoding/json"
 	"fmt"
 	"path/filepath"
 	"reflect"
@@ -232,14 +231,7 @@ func (rts *registryTestSuite) setUp(t *testing.T) func() {
 		t,
 		[]logpb.Channel{logpb.Channel_OPS},
 		[]string{"status_change"},
-		func(entry logpb.Entry) (eventpb.StatusChange, error) {
-			var structuredPayload eventpb.StatusChange
-			err := json.Unmarshal([]byte(entry.Message[entry.StructuredStart:entry.StructuredEnd]), &structuredPayload)
-			if err != nil {
-				return structuredPayload, err
-			}
-			return structuredPayload, nil
-		},
+		logtestutils.FromLogEntry[eventpb.StatusChange],
 	)
 
 	rts.statusChangeLogSpy = spy

pkg/sql/event_log_test.go

@@ -125,13 +125,7 @@ func TestStructuredEventLogging_txnReadTimestamp(t *testing.T) {
 		t,
 		[]logpb.Channel{logpb.Channel_SQL_SCHEMA},
 		[]string{"create_table"},
-		func(entry logpb.Entry) (eventpb.CreateTable, error) {
-			var cte eventpb.CreateTable
-			if err := json.Unmarshal([]byte(entry.Message[entry.StructuredStart:entry.StructuredEnd]), &cte); err != nil {
-				return cte, err
-			}
-			return cte, nil
-		},
+		logtestutils.FromLogEntry[eventpb.CreateTable],
 	)
 
 	cleanup := log.InterceptWith(ctx, appLogsSpy)

pkg/sql/sql_exec_log_test.go

@@ -7,7 +7,6 @@ package sql
 
 import (
 	"context"
-	"encoding/json"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
@@ -38,13 +37,7 @@ func TestSqlExecLog(t *testing.T) {
 		t,
 		[]logpb.Channel{logpb.Channel_SQL_EXEC},
 		[]string{"query_execute"},
-		func(entry logpb.Entry) (eventpb.QueryExecute, error) {
-			var qe eventpb.QueryExecute
-			if err := json.Unmarshal([]byte(entry.Message[entry.StructuredStart:entry.StructuredEnd]), &qe); err != nil {
-				return qe, err
-			}
-			return qe, nil
-		},
+		logtestutils.FromLogEntry[eventpb.QueryExecute],
 		func(entry logpb.Entry, qe eventpb.QueryExecute) bool {
 			// Filter out internal queries.
 			return qe.ExecMode != executorTypeInternal.logLabel()

pkg/util/log/event_log_test.go

@@ -7,7 +7,6 @@ package log_test
 
 import (
 	"context"
-	"encoding/json"
 	"testing"
 	"time"
 
@@ -73,12 +72,8 @@ func TestEventLog(t *testing.T) {
 		[]logpb.Channel{logpb.Channel_DEV},
 		[]string{logtestutils.TestEventType},
 		func(entry logpb.Entry) (TestEventSev, error) {
-			var structuredPayload logtestutils.TestEvent
-			err := json.Unmarshal([]byte(entry.Message[entry.StructuredStart:entry.StructuredEnd]), &structuredPayload)
-			if err != nil {
-				return TestEventSev{event: structuredPayload, severity: entry.Severity}, err
-			}
-			return TestEventSev{event: structuredPayload, severity: entry.Severity}, nil
+			te, err := logtestutils.FromLogEntry[logtestutils.TestEvent](entry)
+			return TestEventSev{event: te, severity: entry.Severity}, err
 		},
 	)
 

pkg/util/log/logtestutils/structured_log_spy.go

@@ -104,6 +104,12 @@ func FormatEntryAsJSON(entry logpb.Entry) (string, error) {
 	return printJSONMap(jsonMap)
 }
 
+func FromLogEntry[T any](entry logpb.Entry) (T, error) {
+	var payload T
+	err := json.Unmarshal([]byte(entry.Message[entry.StructuredStart:entry.StructuredEnd]), &payload)
+	return payload, err
+}
+
 // StructuredLogSpy is a test utility that intercepts structured log entries
 // and stores them in memory. It can be used to verify the contents of log
 // entries in tests.
@@ -126,6 +132,9 @@ type StructuredLogSpy[T any] struct {
 	// that match the event type in the log message.
 	eventTypeRe []*regexp.Regexp
 
+	// Function to transform log entries into the desired format.
+	format func(entry logpb.Entry) (T, error)
+
 	mu struct {
 		syncutil.RWMutex
 
@@ -135,9 +144,6 @@ type StructuredLogSpy[T any] struct {
 		// to determine if the log should be intercepted.
 		filters []func(entry logpb.Entry, formattedEntry T) bool
 
-		// Function to transform log entries into the desired format.
-		format func(entry logpb.Entry) (T, error)
-
 		// lastReadIdx is a map of channel to int, representing the last read log
 		// line read when calling GetUnreadLogs.
 		lastReadIdx map[logpb.Channel]int
@@ -161,6 +167,7 @@ func NewStructuredLogSpy[T any](
 	filters ...func(entry logpb.Entry, formattedEntry T) bool,
 ) *StructuredLogSpy[T] {
 	s := &StructuredLogSpy[T]{
+		format:    format,
 		testState: testState,
 		channels:  make(map[logpb.Channel]struct{}, len(channels)),
 	}
@@ -170,7 +177,6 @@ func NewStructuredLogSpy[T any](
 	}
 	s.mu.lastReadIdx = make(map[logpb.Channel]int, len(channels))
 	s.mu.logs = make(map[logpb.Channel][]T, len(s.channels))
-	s.mu.format = format
 	s.mu.filters = append(s.mu.filters, filters...)
 	s.eventTypes = eventTypes
 	for _, eventType := range eventTypes {
@@ -287,21 +293,21 @@ func (s *StructuredLogSpy[T]) Intercept(entry []byte) {
 		}
 	}
 
-	formattedLog, err := s.mu.format(logEntry)
+	formattedLog, err := s.format(logEntry)
 	if err != nil {
 		s.testState.Fatal(err)
 	}
 
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	if s.mu.filters != nil {
 		for _, filter := range s.mu.filters {
 			if !filter(logEntry, formattedLog) {
 				return
 			}
 		}
 	}
-
-	s.mu.Lock()
-	defer s.mu.Unlock()
 	s.mu.logs[logEntry.Channel] = append(s.mu.logs[logEntry.Channel], formattedLog)
 }