GODRIVER-2943 Re-organize client entity expected log messages (#1366)

(cherry picked from commit 2f0917c)

mongo/integration/unified/client_entity.go

@@ -107,7 +107,10 @@ func newClientEntity(ctx context.Context, em *EntityMap, entityOptions *entityOp
 	}
 
 	if olm := entityOptions.ObserveLogMessages; olm != nil {
-		clientLogger := newLogger(olm, expectedLogMessageCount(ctx))
+		expectedLogMessagesCount := expectedLogMessagesCount(ctx, entityOptions.ID)
+		ignoreLogMessages := ignoreLogMessages(ctx, entityOptions.ID)
+
+		clientLogger := newLogger(olm, expectedLogMessagesCount, ignoreLogMessages)
 
 		wrap := func(str string) options.LogLevel {
 			return options.LogLevel(logger.ParseLevel(str))

mongo/integration/unified/context.go

@@ -28,23 +28,23 @@ const (
 	failPointsKey ctxKey = "test-failpoints"
 	// targetedFailPointsKey is used to store a map from a fail point name to the host on which the fail point is set.
 	targetedFailPointsKey ctxKey = "test-targeted-failpoints"
-	// expectedLogMessageCountKey is used to store the number of log messages expected to be received by the test runner.
-	expectedLogMessageCountKey ctxKey = "test-expected-log-message-count"
+	clientLogMessagesKey  ctxKey = "test-expected-log-message-count"
+	ignoreLogMessagesKey  ctxKey = "test-ignore-log-message-count"
 )
 
 // newTestContext creates a new Context derived from ctx with values initialized to store the state required for test
 // execution.
 func newTestContext(
 	ctx context.Context,
 	entityMap *EntityMap,
-	expectedLogMessageCount int,
+	clientLogMessages []*clientLogMessages,
 	hasOperationalFailPoint bool,
 ) context.Context {
 	ctx = context.WithValue(ctx, operationalFailPointKey, hasOperationalFailPoint)
 	ctx = context.WithValue(ctx, entitiesKey, entityMap)
 	ctx = context.WithValue(ctx, failPointsKey, make(map[string]*mongo.Client))
 	ctx = context.WithValue(ctx, targetedFailPointsKey, make(map[string]string))
-	ctx = context.WithValue(ctx, expectedLogMessageCountKey, expectedLogMessageCount)
+	ctx = context.WithValue(ctx, clientLogMessagesKey, clientLogMessages)
 	return ctx
 }
 
@@ -84,6 +84,28 @@ func entities(ctx context.Context) *EntityMap {
 	return ctx.Value(entitiesKey).(*EntityMap)
 }
 
-func expectedLogMessageCount(ctx context.Context) int {
-	return ctx.Value(expectedLogMessageCountKey).(int)
+func expectedLogMessagesCount(ctx context.Context, clientID string) int {
+	messages := ctx.Value(clientLogMessagesKey).([]*clientLogMessages)
+
+	count := 0
+	for _, message := range messages {
+		if message.Client == clientID {
+			count += len(message.LogMessages)
+		}
+	}
+
+	return count
+}
+
+func ignoreLogMessages(ctx context.Context, clientID string) []*logMessage {
+	messages := ctx.Value(clientLogMessagesKey).([]*clientLogMessages)
+
+	ignoreMessages := []*logMessage{}
+	for _, message := range messages {
+		if message.Client == clientID {
+			ignoreMessages = append(ignoreMessages, message.IgnoreMessages...)
+		}
+	}
+
+	return ignoreMessages
 }

mongo/integration/unified/logger.go

@@ -7,6 +7,7 @@
 package unified
 
 import (
+	"context"
 	"sync"
 
 	"go.mongodb.org/mongo-driver/internal/logger"
@@ -33,19 +34,21 @@ type Logger struct {
 	// orderMu guards the order value, which increments each time the "Info"
 	// method is called. This is necessary since "Info" could be called from
 	// multiple go routines, e.g. SDAM logs.
-	orderMu  sync.RWMutex
-	logQueue chan orderedLogMessage
+	orderMu        sync.RWMutex
+	logQueue       chan orderedLogMessage
+	ignoreMessages []*logMessage
 }
 
-func newLogger(olm *observeLogMessages, bufSize int) *Logger {
+func newLogger(olm *observeLogMessages, bufSize int, ignoreMessages []*logMessage) *Logger {
 	if olm == nil {
 		return nil
 	}
 
 	return &Logger{
-		lastOrder: 1,
-		logQueue:  make(chan orderedLogMessage, bufSize),
-		bufSize:   bufSize,
+		lastOrder:      1,
+		logQueue:       make(chan orderedLogMessage, bufSize),
+		bufSize:        bufSize,
+		ignoreMessages: ignoreMessages,
 	}
 }
 
@@ -65,8 +68,6 @@ func (log *Logger) Info(level int, msg string, args ...interface{}) {
 		return
 	}
 
-	defer func() { log.lastOrder++ }()
-
 	// Add the Diff back to the level, as there is no need to create a
 	// logging offset.
 	level = level + logger.DiffToInfo
@@ -76,12 +77,19 @@ func (log *Logger) Info(level int, msg string, args ...interface{}) {
 		panic(err)
 	}
 
+	for _, ignoreMessage := range log.ignoreMessages {
+		if err := verifyLogMatch(context.Background(), ignoreMessage, logMessage); err == nil {
+			return
+		}
+	}
+
+	defer func() { log.lastOrder++ }()
+
 	// Send the log message to the "orderedLogMessage" channel for
 	// validation.
 	log.logQueue <- orderedLogMessage{
 		order:      log.lastOrder + 1,
-		logMessage: logMessage,
-	}
+		logMessage: logMessage}
 
 	// If the order has reached the buffer size, then close the channel.
 	if log.lastOrder == log.bufSize {

mongo/integration/unified/logger_verification.go

@@ -79,17 +79,6 @@ type clientLogMessages struct {
 	LogMessages    []*logMessage `bson:"messages"`
 }
 
-// ignore checks to see if the message is in the "IgnoreMessages" slice.
-func (clm clientLogMessages) ignore(ctx context.Context, msg *logMessage) bool {
-	for _, ignoreMessage := range clm.IgnoreMessages {
-		if err := verifyLogMatch(ctx, ignoreMessage, msg); err == nil {
-			return true
-		}
-	}
-
-	return false
-}
-
 // logMessageValidator defines the expectation for log messages across all
 // clients.
 type logMessageValidator struct {
@@ -191,8 +180,7 @@ type logQueues struct {
 }
 
 // partitionLogQueue will partition the expected logs into "unordered" and
-// "ordered" log channels. This function will also remove any logs in the
-// "ignoreMessages" list for a client.
+// "ordered" log channels.
 func partitionLogQueue(ctx context.Context, exp *clientLogMessages) logQueues {
 	orderedLogCh := make(chan *logMessage, len(exp.LogMessages))
 	unorderedLogCh := make(chan *logMessage, len(exp.LogMessages))
@@ -241,12 +229,6 @@ func matchOrderedLogs(ctx context.Context, logs logQueues) <-chan error {
 		defer close(errs)
 
 		for actual := range logs.ordered {
-			// Ignore logs that are in the "IngoreMessages" slice of
-			// the expected results.
-			if logs.expected.ignore(ctx, actual) {
-				continue
-			}
-
 			expected := expLogMessages[0]
 			if expected == nil {
 				continue

mongo/integration/unified/unified_spec_runner.go

@@ -30,8 +30,6 @@ var (
 		// TODO(GODRIVER-2843): Fix and unskip these test cases.
 		"Find operation with snapshot":                                      "Test fails frequently. See GODRIVER-2843",
 		"Write commands with snapshot session do not affect snapshot reads": "Test fails frequently. See GODRIVER-2843",
-		// TODO(GODRIVER-2943): Fix and unskip this test case.
-		"Topology lifecycle": "Test times out.  See GODRIVER-2943",
 	}
 
 	logMessageValidatorTimeout = 10 * time.Millisecond
@@ -232,13 +230,7 @@ func (tc *TestCase) Run(ls LoggerSkipper) error {
 		return fmt.Errorf("schema version %q not supported: %v", tc.schemaVersion, err)
 	}
 
-	// Count the number of expected log messages over all clients.
-	var expectedLogCount int
-	for _, clientLog := range tc.ExpectLogMessages {
-		expectedLogCount += len(clientLog.LogMessages)
-	}
-
-	testCtx := newTestContext(context.Background(), tc.entities, expectedLogCount, tc.setsFailPoint())
+	testCtx := newTestContext(context.Background(), tc.entities, tc.ExpectLogMessages, tc.setsFailPoint())
 
 	defer func() {
 		// If anything fails while doing test cleanup, we only log the error because the actual test may have already

testdata/server-discovery-and-monitoring/unified/logging-replicaset.json

@@ -220,22 +220,6 @@
                 }
               }
             },
-            {
-              "level": "debug",
-              "component": "topology",
-              "data": {
-                "message": "Topology description changed",
-                "topologyId": {
-                  "$$exists": true
-                },
-                "previousDescription": {
-                  "$$exists": true
-                },
-                "newDescription": {
-                  "$$exists": true
-                }
-              }
-            },
             {
               "level": "debug",
               "component": "topology",

testdata/server-discovery-and-monitoring/unified/logging-replicaset.yml

@@ -109,6 +109,13 @@ tests:
               topologyId: { $$exists: true }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
+          # TODO(GODRIVER-2967): The following log message has been removed from
+          # the JSON analogue because it assumes that 
+          # "TopologyDescriptionChangedEvent" should occur when a topolgoy is 
+          # closed. This behavior is not clearly defined anywhere and some 
+          # drivers support and some don't.
+          #
+          # Need to sync whenever GODRIVER-2967 is unblocked.
           - level: debug
             component: topology
             data:

testdata/server-discovery-and-monitoring/unified/logging-sharded.json

@@ -187,22 +187,6 @@
                 }
               }
             },
-            {
-              "level": "debug",
-              "component": "topology",
-              "data": {
-                "message": "Topology description changed",
-                "topologyId": {
-                  "$$exists": true
-                },
-                "previousDescription": {
-                  "$$exists": true
-                },
-                "newDescription": {
-                  "$$exists": true
-                }
-              }
-            },
             {
               "level": "debug",
               "component": "topology",

testdata/server-discovery-and-monitoring/unified/logging-sharded.yml

@@ -97,6 +97,13 @@ tests:
               topologyId: { $$exists: true }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
+          # TODO(GODRIVER-2967): The following log message has been removed from
+          # the JSON analogue because it assumes that 
+          # "TopologyDescriptionChangedEvent" should occur when a topolgoy is 
+          # closed. This behavior is not clearly defined anywhere and some 
+          # drivers support and some don't.
+          #
+          # Need to sync whenever GODRIVER-2967 is unblocked.
           - level: debug
             component: topology
             data:

testdata/server-discovery-and-monitoring/unified/logging-standalone.json

@@ -162,22 +162,6 @@
                 }
               }
             },
-            {
-              "level": "debug",
-              "component": "topology",
-              "data": {
-                "message": "Topology description changed",
-                "topologyId": {
-                  "$$exists": true
-                },
-                "previousDescription": {
-                  "$$exists": true
-                },
-                "newDescription": {
-                  "$$exists": true
-                }
-              }
-            },
             {
               "level": "debug",
               "component": "topology",

testdata/server-discovery-and-monitoring/unified/logging-standalone.yml

@@ -84,6 +84,13 @@ tests:
               topologyId: { $$exists: true }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
+          # TODO(GODRIVER-2967): The following log message has been removed from
+          # the JSON analogue because it assumes that 
+          # "TopologyDescriptionChangedEvent" should occur when a topolgoy is 
+          # closed. This behavior is not clearly defined anywhere and some 
+          # drivers support and some don't.
+          #
+          # Need to sync whenever GODRIVER-2967 is unblocked.
           - level: debug
             component: topology
             data: