Added logging specification tests

driver-core/src/main/com/mongodb/internal/connection/BaseCluster.java

@@ -118,9 +118,9 @@ abstract class BaseCluster implements Cluster {
         this.clusterListener = singleClusterListener(settings);
         ClusterOpeningEvent clusterOpeningEvent = new ClusterOpeningEvent(clusterId);
         clusterListener.clusterOpening(clusterOpeningEvent);
-        logTopologyOpening(clusterId, clusterOpeningEvent);
         description = new ClusterDescription(settings.getMode(), UNKNOWN, emptyList(),
                 settings, serverFactory.getSettings());
+        logTopologyMonitoringStarting(clusterId);
     }
 
     @Override
@@ -221,7 +221,7 @@ public void close() {
                     description);
             ClusterClosedEvent clusterClosedEvent = new ClusterClosedEvent(clusterId);
             clusterListener.clusterClosed(clusterClosedEvent);
-            logTopologyClosedEvent(clusterId, clusterClosedEvent);
+            logTopologyMonitoringStopping(clusterId);
             stopWaitQueueHandler();
         }
     }
@@ -632,9 +632,7 @@ static void logServerSelectionSucceeded(
         }
     }
 
-    static void logTopologyOpening(
-            final ClusterId clusterId,
-            final ClusterOpeningEvent clusterOpeningEvent) {
+    static void logTopologyMonitoringStarting(final ClusterId clusterId) {
         if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
             STRUCTURED_LOGGER.log(new LogMessage(
                     TOPOLOGY, DEBUG, "Starting topology monitoring", clusterId,
@@ -659,9 +657,7 @@ static void logTopologyDescriptionChanged(
         }
     }
 
-    static void logTopologyClosedEvent(
-            final ClusterId clusterId,
-            final ClusterClosedEvent clusterClosedEvent) {
+    static void logTopologyMonitoringStopping(final ClusterId clusterId) {
         if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
             STRUCTURED_LOGGER.log(new LogMessage(
                     TOPOLOGY, DEBUG, "Stopped topology monitoring", clusterId,

driver-core/src/main/com/mongodb/internal/connection/DefaultServer.java

@@ -31,6 +31,7 @@
 import com.mongodb.internal.connection.SdamServerDescriptionManager.SdamIssue;
 import com.mongodb.internal.diagnostics.logging.Logger;
 import com.mongodb.internal.diagnostics.logging.Loggers;
+import com.mongodb.internal.logging.StructuredLogger;
 import com.mongodb.internal.session.SessionContext;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonDocument;
@@ -48,6 +49,7 @@
 
 class DefaultServer implements ClusterableServer {
     private static final Logger LOGGER = Loggers.getLogger("connection");
+    private static final StructuredLogger STRUCTURED_LOGGER = new StructuredLogger("connection");
     private final ServerId serverId;
     private final ConnectionPool connectionPool;
     private final ClusterConnectionMode clusterConnectionMode;
@@ -75,7 +77,6 @@ class DefaultServer implements ClusterableServer {
         this.connectionPool = notNull("connectionPool", connectionPool);
 
         this.serverId = serverId;
-
         serverListener.serverOpening(new ServerOpeningEvent(this.serverId));
 
         this.serverMonitor = serverMonitor;

driver-core/src/main/com/mongodb/internal/connection/DefaultServerMonitor.java

@@ -22,6 +22,7 @@
 import com.mongodb.ServerApi;
 import com.mongodb.annotations.ThreadSafe;
 import com.mongodb.connection.ClusterConnectionMode;
+import com.mongodb.connection.ConnectionDescription;
 import com.mongodb.connection.ServerDescription;
 import com.mongodb.connection.ServerId;
 import com.mongodb.connection.ServerSettings;
@@ -34,6 +35,8 @@
 import com.mongodb.internal.diagnostics.logging.Logger;
 import com.mongodb.internal.diagnostics.logging.Loggers;
 import com.mongodb.internal.inject.Provider;
+import com.mongodb.internal.logging.LogMessage;
+import com.mongodb.internal.logging.StructuredLogger;
 import com.mongodb.internal.validator.NoOpFieldNameValidator;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonBoolean;
@@ -63,14 +66,27 @@
 import static com.mongodb.internal.connection.DescriptionHelper.createServerDescription;
 import static com.mongodb.internal.connection.ServerDescriptionHelper.unknownConnectingServerDescription;
 import static com.mongodb.internal.event.EventListenerHelper.singleServerMonitorListener;
+import static com.mongodb.internal.logging.LogMessage.Component.TOPOLOGY;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.AWAITED;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.DRIVER_CONNECTION_ID;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.DURATION_MS;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.FAILURE;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.REPLY;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_CONNECTION_ID;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_HOST;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_PORT;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_ID;
+import static com.mongodb.internal.logging.LogMessage.Level.DEBUG;
 import static java.lang.String.format;
+import static java.util.Arrays.asList;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
 @ThreadSafe
 class DefaultServerMonitor implements ServerMonitor {
 
     private static final Logger LOGGER = Loggers.getLogger("cluster");
+    private static final StructuredLogger STRUCTURED_LOGGER = new StructuredLogger("cluster");
 
     private final ServerId serverId;
     private final ServerMonitorListener serverMonitorListener;
@@ -116,6 +132,7 @@ class DefaultServerMonitor implements ServerMonitor {
 
     @Override
     public void start() {
+        logStartedServerMonitoring(serverId);
         monitor.start();
     }
 
@@ -137,6 +154,9 @@ public void connect() {
     @SuppressWarnings("try")
     public void close() {
         withLock(lock, () -> {
+            if (!isClosed) {
+                logStoppedServerMonitoring(serverId);
+            }
             isClosed = true;
             //noinspection EmptyTryBlock
             try (ServerMonitor ignoredAutoClosed = monitor;
@@ -226,6 +246,7 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                     alreadyLoggedHeartBeatStarted = true;
                     currentCheckCancelled = false;
                     InternalConnection newConnection = internalConnectionFactory.create(serverId);
+                    logHeartbeatStarted(serverId, newConnection.getDescription(), shouldStreamResponses);
                     serverMonitorListener.serverHearbeatStarted(new ServerHeartbeatStartedEvent(
                             newConnection.getDescription().getConnectionId(), shouldStreamResponses));
                     newConnection.open(operationContextFactory.create());
@@ -238,6 +259,7 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                     LOGGER.debug(format("Checking status of %s", serverId.getAddress()));
                 }
                 if (!alreadyLoggedHeartBeatStarted) {
+                    logHeartbeatStarted(serverId, connection.getDescription(), shouldStreamResponses);
                     serverMonitorListener.serverHearbeatStarted(new ServerHeartbeatStartedEvent(
                             connection.getDescription().getConnectionId(), shouldStreamResponses));
                 }
@@ -269,15 +291,18 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                     if (!shouldStreamResponses) {
                         roundTripTimeSampler.addSample(elapsedTimeNanos);
                     }
+                    logHeartbeatSucceeded(serverId, connection.getDescription(), shouldStreamResponses, elapsedTimeNanos, helloResult);
                     serverMonitorListener.serverHeartbeatSucceeded(
                             new ServerHeartbeatSucceededEvent(connection.getDescription().getConnectionId(), helloResult,
                                     elapsedTimeNanos, shouldStreamResponses));
 
                     return createServerDescription(serverId.getAddress(), helloResult, roundTripTimeSampler.getAverage(),
                             roundTripTimeSampler.getMin());
                 } catch (Exception e) {
+                    long elapsedTimeNanos = System.nanoTime() - start;
+                    logHeartbeatFailed(serverId, connection.getDescription(), shouldStreamResponses, elapsedTimeNanos, e);
                     serverMonitorListener.serverHeartbeatFailed(
-                            new ServerHeartbeatFailedEvent(connection.getDescription().getConnectionId(), System.nanoTime() - start,
+                            new ServerHeartbeatFailedEvent(connection.getDescription().getConnectionId(), elapsedTimeNanos,
                                     shouldStreamResponses, e));
                     throw e;
                 }
@@ -515,4 +540,94 @@ private void waitForNext() throws InterruptedException {
     private String getHandshakeCommandName(final ServerDescription serverDescription) {
         return serverDescription.isHelloOk() ? HELLO : LEGACY_HELLO;
     }
+
+    private static void logHeartbeatStarted(
+            final ServerId serverId,
+            final ConnectionDescription connectionDescription,
+            final boolean awaited) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, serverId.getClusterId())) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Server heartbeat started", serverId.getClusterId(),
+                    asList(
+                            new LogMessage.Entry(SERVER_HOST, serverId.getAddress().getHost()),
+                            new LogMessage.Entry(SERVER_PORT, serverId.getAddress().getPort()),
+                            new LogMessage.Entry(DRIVER_CONNECTION_ID, connectionDescription.getConnectionId().getLocalValue()),
+                            new LogMessage.Entry(SERVER_CONNECTION_ID, connectionDescription.getConnectionId().getServerValue()),
+                            new LogMessage.Entry(TOPOLOGY_ID, serverId.getClusterId()),
+                            new LogMessage.Entry(AWAITED, awaited)),
+                    "Heartbeat started for {}:{} on connection with driver-generated ID {} and server-generated ID {} "
+                           + "in topology with ID {}. Awaited: {}"));
+        }
+    }
+
+    private static void logHeartbeatSucceeded(
+            final ServerId serverId,
+            final ConnectionDescription connectionDescription,
+            final boolean awaited,
+            final long elapsedTimeNanos,
+            final BsonDocument reply) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, serverId.getClusterId())) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Server heartbeat succeeded", serverId.getClusterId(),
+                    asList(
+                            new LogMessage.Entry(DURATION_MS, MILLISECONDS.convert(elapsedTimeNanos, NANOSECONDS)),
+                            new LogMessage.Entry(SERVER_HOST, serverId.getAddress().getHost()),
+                            new LogMessage.Entry(SERVER_PORT, serverId.getAddress().getPort()),
+                            new LogMessage.Entry(DRIVER_CONNECTION_ID, connectionDescription.getConnectionId().getLocalValue()),
+                            new LogMessage.Entry(SERVER_CONNECTION_ID, connectionDescription.getConnectionId().getServerValue()),
+                            new LogMessage.Entry(TOPOLOGY_ID, serverId.getClusterId()),
+                            new LogMessage.Entry(AWAITED, awaited),
+                            new LogMessage.Entry(REPLY, reply.toJson())),
+                    "Heartbeat succeeded in {} ms for {}:{} on connection with driver-generated ID {} and server-generated ID {} "
+                            + "in topology with ID {}. Awaited: {}. Reply: {}"));
+        }
+    }
+
+    private static void logHeartbeatFailed(
+            final ServerId serverId,
+            final ConnectionDescription connectionDescription,
+            final boolean awaited,
+            final long elapsedTimeNanos,
+            final Exception failure) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, serverId.getClusterId())) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Server heartbeat failed", serverId.getClusterId(),
+                    asList(
+                            new LogMessage.Entry(DURATION_MS, MILLISECONDS.convert(elapsedTimeNanos, NANOSECONDS)),
+                            new LogMessage.Entry(SERVER_HOST, serverId.getAddress().getHost()),
+                            new LogMessage.Entry(SERVER_PORT, serverId.getAddress().getPort()),
+                            new LogMessage.Entry(DRIVER_CONNECTION_ID, connectionDescription.getConnectionId().getLocalValue()),
+                            new LogMessage.Entry(SERVER_CONNECTION_ID, connectionDescription.getConnectionId().getServerValue()),
+                            new LogMessage.Entry(TOPOLOGY_ID, serverId.getClusterId()),
+                            new LogMessage.Entry(AWAITED, awaited),
+                            new LogMessage.Entry(FAILURE, failure.getMessage())),
+                    "Heartbeat failed in {} ms for {}:{} on connection with driver-generated ID {} and server-generated ID {} "
+                            + "in topology with ID {}. Awaited: {}. Failure: {}"));
+        }
+    }
+
+
+    private static void logStartedServerMonitoring(final ServerId serverId) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, serverId.getClusterId())) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Starting server monitoring", serverId.getClusterId(),
+                    asList(
+                            new LogMessage.Entry(SERVER_HOST, serverId.getAddress().getHost()),
+                            new LogMessage.Entry(SERVER_PORT, serverId.getAddress().getPort()),
+                            new LogMessage.Entry(TOPOLOGY_ID, serverId.getClusterId())),
+                    "Starting monitoring for server {}:{} in topology with ID {}"));
+        }
+    }
+
+    private static void logStoppedServerMonitoring(final ServerId serverId) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, serverId.getClusterId())) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Stopped server monitoring", serverId.getClusterId(),
+                    asList(
+                            new LogMessage.Entry(SERVER_HOST, serverId.getAddress().getHost()),
+                            new LogMessage.Entry(SERVER_PORT, serverId.getAddress().getPort()),
+                            new LogMessage.Entry(TOPOLOGY_ID, serverId.getClusterId())),
+                    "Stopped monitoring for server {}:{} in topology with ID {}"));
+        }
+    }
 }

driver-core/src/main/com/mongodb/internal/connection/LoadBalancedCluster.java

@@ -63,7 +63,7 @@
 import static com.mongodb.connection.ServerConnectionState.CONNECTING;
 import static com.mongodb.internal.connection.BaseCluster.logServerSelectionStarted;
 import static com.mongodb.internal.connection.BaseCluster.logServerSelectionSucceeded;
-import static com.mongodb.internal.connection.BaseCluster.logTopologyClosedEvent;
+import static com.mongodb.internal.connection.BaseCluster.logTopologyMonitoringStopping;
 import static com.mongodb.internal.event.EventListenerHelper.singleClusterListener;
 import static java.lang.String.format;
 import static java.util.Collections.emptyList;
@@ -275,7 +275,7 @@ public void close() {
             }
             ClusterClosedEvent clusterClosedEvent = new ClusterClosedEvent(clusterId);
             clusterListener.clusterClosed(clusterClosedEvent);
-            logTopologyClosedEvent(clusterId, clusterClosedEvent);
+            logTopologyMonitoringStopping(clusterId);
         }
     }
 

driver-core/src/main/com/mongodb/internal/connection/SingleServerCluster.java

@@ -58,9 +58,9 @@ public SingleServerCluster(final ClusterId clusterId, final ClusterSettings sett
         // synchronized in the constructor because the change listener is re-entrant to this instance.
         // In other words, we are leaking a reference to "this" from the constructor.
         withLock(() -> {
-            server.set(createServer(settings.getHosts().get(0)));
             publishDescription(ServerDescription.builder().state(CONNECTING).address(settings.getHosts().get(0))
                     .build());
+            server.set(createServer(settings.getHosts().get(0)));
         });
     }
 

driver-core/src/main/com/mongodb/internal/logging/LogMessage.java

@@ -108,6 +108,7 @@ public enum Name {
              * Not supported.
              */
             OPERATION("operation"),
+            AWAITED("awaited"),
             SERVICE_ID("serviceId"),
             SERVER_CONNECTION_ID("serverConnectionId"),
             DRIVER_CONNECTION_ID("driverConnectionId"),

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/unified/UnifiedServerDiscoveryAndMonitoringTest.java

@@ -17,7 +17,6 @@
 package com.mongodb.reactivestreams.client.unified;
 
 import org.junit.jupiter.params.provider.Arguments;
-
 import java.util.Collection;
 
 final class UnifiedServerDiscoveryAndMonitoringTest extends UnifiedReactiveStreamsTest {

driver-sync/src/test/functional/com/mongodb/client/unified/ContextElement.java

@@ -495,7 +495,7 @@ public String toString() {
                     + new BsonDocument("messages", expectedMessages).toJson(JsonWriterSettings.builder().indent(true).build()) + "\n"
                     + "   actualMessages="
                     + new BsonDocument("messages", new BsonArray(actualMessages.stream()
-                    .map(LogMatcher::asDocument).collect(Collectors.toList())))
+                    .map(LogMatcher::logMessageAsDocument).collect(Collectors.toList())))
                     .toJson(JsonWriterSettings.builder().indent(true).build()) + "\n";
         }
     }

driver-sync/src/test/functional/com/mongodb/client/unified/EventMatcher.java

@@ -336,7 +336,7 @@ public <T> void waitForServerMonitorEvents(final String client, final Class<T> e
         BsonDocument expectedEventContents = getEventContents(expectedEvent);
         try {
             serverMonitorListener.waitForEvents(expectedEventType,
-                    event -> serverMonitorEventMatches(expectedEventContents, event, null), count, Duration.ofSeconds(10));
+                    event -> serverMonitorEventMatches(expectedEventContents, event, null), count, Duration.ofSeconds(15));
             context.pop();
         } catch (InterruptedException e) {
             throw new RuntimeException(e);