Skip to content

Commit a20d204

Browse files
rozzaNathanQingyangXu
rozza
and
NathanQingyangXu
authored
Publish ClusterDescriptionChangedEvent on topology close (#1738)
As well as add Topology events to structured logging JAVA-5229 --------- Co-authored-by: Nathan Xu <[email protected]>
1 parent 47a0717 commit a20d204

File tree

12 files changed

+286
-27
lines changed

12 files changed

+286
-27
lines changed

‎driver-core/src/main/com/mongodb/event/ClusterClosedEvent.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
/**
2424
* A cluster closed event.
2525
*
26+
* <p>This event is synonymous with TopologyClosedEvent</p>
27+
*
2628
* @since 3.3
2729
*/
2830
public final class ClusterClosedEvent {

‎driver-core/src/main/com/mongodb/event/ClusterDescriptionChangedEvent.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
/**
2525
* An event signifying that the cluster description has changed.
2626
*
27+
* <p>This event is synonymous with TopologyDescriptionChangedEvent</p>
28+
*
2729
* @since 3.3
2830
*/
2931
public final class ClusterDescriptionChangedEvent {

‎driver-core/src/main/com/mongodb/event/ClusterOpeningEvent.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
/**
2424
* A cluster opening event.
2525
*
26+
* <p>This event is synonymous with TopologyOpeningEvent</p>
27+
*
2628
* @since 3.3
2729
*/
2830
public final class ClusterOpeningEvent {

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

Lines changed: 58 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@
2727
import com.mongodb.connection.ClusterDescription;
2828
import com.mongodb.connection.ClusterId;
2929
import com.mongodb.connection.ClusterSettings;
30-
import com.mongodb.connection.ClusterType;
3130
import com.mongodb.connection.ServerDescription;
3231
import com.mongodb.event.ClusterClosedEvent;
3332
import com.mongodb.event.ClusterDescriptionChangedEvent;
@@ -50,7 +49,6 @@
5049
import com.mongodb.selector.CompositeServerSelector;
5150
import com.mongodb.selector.ServerSelector;
5251

53-
import java.util.Collections;
5452
import java.util.Deque;
5553
import java.util.Iterator;
5654
import java.util.List;
@@ -64,6 +62,7 @@
6462
import static com.mongodb.assertions.Assertions.assertNotNull;
6563
import static com.mongodb.assertions.Assertions.isTrue;
6664
import static com.mongodb.assertions.Assertions.notNull;
65+
import static com.mongodb.connection.ClusterType.UNKNOWN;
6766
import static com.mongodb.connection.ServerDescription.MAX_DRIVER_WIRE_VERSION;
6867
import static com.mongodb.connection.ServerDescription.MIN_DRIVER_SERVER_VERSION;
6968
import static com.mongodb.connection.ServerDescription.MIN_DRIVER_WIRE_VERSION;
@@ -72,6 +71,7 @@
7271
import static com.mongodb.internal.connection.EventHelper.wouldDescriptionsGenerateEquivalentEvents;
7372
import static com.mongodb.internal.event.EventListenerHelper.singleClusterListener;
7473
import static com.mongodb.internal.logging.LogMessage.Component.SERVER_SELECTION;
74+
import static com.mongodb.internal.logging.LogMessage.Component.TOPOLOGY;
7575
import static com.mongodb.internal.logging.LogMessage.Entry.Name.FAILURE;
7676
import static com.mongodb.internal.logging.LogMessage.Entry.Name.OPERATION;
7777
import static com.mongodb.internal.logging.LogMessage.Entry.Name.OPERATION_ID;
@@ -80,11 +80,16 @@
8080
import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_HOST;
8181
import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_PORT;
8282
import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_DESCRIPTION;
83+
import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_ID;
84+
import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_NEW_DESCRIPTION;
85+
import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_PREVIOUS_DESCRIPTION;
8386
import static com.mongodb.internal.logging.LogMessage.Level.DEBUG;
8487
import static com.mongodb.internal.logging.LogMessage.Level.INFO;
8588
import static com.mongodb.internal.time.Timeout.ZeroSemantics.ZERO_DURATION_MEANS_EXPIRED;
8689
import static java.lang.String.format;
8790
import static java.util.Arrays.asList;
91+
import static java.util.Collections.emptyList;
92+
import static java.util.Collections.singletonList;
8893
import static java.util.concurrent.TimeUnit.MILLISECONDS;
8994
import static java.util.concurrent.TimeUnit.NANOSECONDS;
9095
import static java.util.stream.Collectors.toList;
@@ -111,8 +116,10 @@ abstract class BaseCluster implements Cluster {
111116
this.settings = notNull("settings", settings);
112117
this.serverFactory = notNull("serverFactory", serverFactory);
113118
this.clusterListener = singleClusterListener(settings);
114-
clusterListener.clusterOpening(new ClusterOpeningEvent(clusterId));
115-
description = new ClusterDescription(settings.getMode(), ClusterType.UNKNOWN, Collections.emptyList(),
119+
ClusterOpeningEvent clusterOpeningEvent = new ClusterOpeningEvent(clusterId);
120+
clusterListener.clusterOpening(clusterOpeningEvent);
121+
logTopologyOpening(clusterId, clusterOpeningEvent);
122+
description = new ClusterDescription(settings.getMode(), UNKNOWN, emptyList(),
116123
settings, serverFactory.getSettings());
117124
}
118125

@@ -210,7 +217,11 @@ public void close() {
210217
if (!isClosed()) {
211218
isClosed = true;
212219
phase.get().countDown();
213-
clusterListener.clusterClosed(new ClusterClosedEvent(clusterId));
220+
fireChangeEvent(new ClusterDescription(settings.getMode(), UNKNOWN, emptyList(), settings, serverFactory.getSettings()),
221+
description);
222+
ClusterClosedEvent clusterClosedEvent = new ClusterClosedEvent(clusterId);
223+
clusterListener.clusterClosed(clusterClosedEvent);
224+
logTopologyClosedEvent(clusterId, clusterClosedEvent);
214225
stopWaitQueueHandler();
215226
}
216227
}
@@ -237,8 +248,9 @@ protected void updateDescription(final ClusterDescription newDescription) {
237248
*/
238249
protected void fireChangeEvent(final ClusterDescription newDescription, final ClusterDescription previousDescription) {
239250
if (!wouldDescriptionsGenerateEquivalentEvents(newDescription, previousDescription)) {
240-
clusterListener.clusterDescriptionChanged(
241-
new ClusterDescriptionChangedEvent(getClusterId(), newDescription, previousDescription));
251+
ClusterDescriptionChangedEvent changedEvent = new ClusterDescriptionChangedEvent(getClusterId(), newDescription, previousDescription);
252+
clusterListener.clusterDescriptionChanged(changedEvent);
253+
logTopologyDescriptionChanged(getClusterId(), changedEvent);
242254
}
243255
}
244256

@@ -619,4 +631,43 @@ static void logServerSelectionSucceeded(
619631
+ " Selector: {}, topology description: {}"));
620632
}
621633
}
634+
635+
static void logTopologyOpening(
636+
final ClusterId clusterId,
637+
final ClusterOpeningEvent clusterOpeningEvent) {
638+
if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
639+
STRUCTURED_LOGGER.log(new LogMessage(
640+
TOPOLOGY, DEBUG, "Starting topology monitoring", clusterId,
641+
singletonList(new Entry(TOPOLOGY_ID, clusterId)),
642+
"Starting monitoring for topology with ID {}"));
643+
}
644+
}
645+
646+
static void logTopologyDescriptionChanged(
647+
final ClusterId clusterId,
648+
final ClusterDescriptionChangedEvent clusterDescriptionChangedEvent) {
649+
if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
650+
STRUCTURED_LOGGER.log(new LogMessage(
651+
TOPOLOGY, DEBUG, "Topology description changed", clusterId,
652+
asList(
653+
new Entry(TOPOLOGY_ID, clusterId),
654+
new Entry(TOPOLOGY_PREVIOUS_DESCRIPTION,
655+
clusterDescriptionChangedEvent.getPreviousDescription().getShortDescription()),
656+
new Entry(TOPOLOGY_NEW_DESCRIPTION,
657+
clusterDescriptionChangedEvent.getNewDescription().getShortDescription())),
658+
"Description changed for topology with ID {}. Previous description: {}. New description: {}"));
659+
}
660+
}
661+
662+
static void logTopologyClosedEvent(
663+
final ClusterId clusterId,
664+
final ClusterClosedEvent clusterClosedEvent) {
665+
if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
666+
STRUCTURED_LOGGER.log(new LogMessage(
667+
TOPOLOGY, DEBUG, "Stopped topology monitoring", clusterId,
668+
singletonList(new Entry(TOPOLOGY_ID, clusterId)),
669+
"Stopped monitoring for topology with ID {}"));
670+
}
671+
}
672+
622673
}

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
import static com.mongodb.connection.ServerConnectionState.CONNECTING;
6464
import static com.mongodb.internal.connection.BaseCluster.logServerSelectionStarted;
6565
import static com.mongodb.internal.connection.BaseCluster.logServerSelectionSucceeded;
66+
import static com.mongodb.internal.connection.BaseCluster.logTopologyClosedEvent;
6667
import static com.mongodb.internal.event.EventListenerHelper.singleClusterListener;
6768
import static java.lang.String.format;
6869
import static java.util.Collections.emptyList;
@@ -272,7 +273,9 @@ public void close() {
272273
if (localServer != null) {
273274
localServer.close();
274275
}
275-
clusterListener.clusterClosed(new ClusterClosedEvent(clusterId));
276+
ClusterClosedEvent clusterClosedEvent = new ClusterClosedEvent(clusterId);
277+
clusterListener.clusterClosed(clusterClosedEvent);
278+
logTopologyClosedEvent(clusterId, clusterClosedEvent);
276279
}
277280
}
278281

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,8 @@ public final class LogMessage {
4848
public enum Component {
4949
COMMAND("command"),
5050
CONNECTION("connection"),
51-
SERVER_SELECTION("serverSelection");
51+
SERVER_SELECTION("serverSelection"),
52+
TOPOLOGY("topology");
5253

5354
private static final Map<String, Component> INDEX;
5455

@@ -124,7 +125,10 @@ public enum Name {
124125
WAIT_QUEUE_TIMEOUT_MS("waitQueueTimeoutMS"),
125126
SELECTOR("selector"),
126127
TOPOLOGY_DESCRIPTION("topologyDescription"),
127-
REMAINING_TIME_MS("remainingTimeMS");
128+
REMAINING_TIME_MS("remainingTimeMS"),
129+
TOPOLOGY_ID("topologyId"),
130+
TOPOLOGY_PREVIOUS_DESCRIPTION("previousDescription"),
131+
TOPOLOGY_NEW_DESCRIPTION("newDescription");
128132

129133
private final String value;
130134

‎driver-core/src/test/unit/com/mongodb/internal/connection/TestClusterListener.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import java.time.Duration;
2626
import java.util.ArrayList;
2727
import java.util.List;
28+
import java.util.concurrent.CountDownLatch;
29+
import java.util.concurrent.TimeUnit;
2830
import java.util.concurrent.TimeoutException;
2931
import java.util.concurrent.locks.Condition;
3032
import java.util.concurrent.locks.ReentrantLock;
@@ -42,6 +44,7 @@ public final class TestClusterListener implements ClusterListener {
4244
private final ArrayList<ClusterDescriptionChangedEvent> clusterDescriptionChangedEvents = new ArrayList<>();
4345
private final ReentrantLock lock = new ReentrantLock();
4446
private final Condition newClusterDescriptionChangedEventCondition = lock.newCondition();
47+
private final CountDownLatch closedLatch = new CountDownLatch(1);
4548

4649
@Override
4750
public void clusterOpening(final ClusterOpeningEvent event) {
@@ -52,6 +55,7 @@ public void clusterOpening(final ClusterOpeningEvent event) {
5255
@Override
5356
public void clusterClosed(final ClusterClosedEvent event) {
5457
isTrue("clusterClosingEvent is null", clusterClosingEvent == null);
58+
closedLatch.countDown();
5559
clusterClosingEvent = event;
5660
}
5761

@@ -110,6 +114,17 @@ public void waitForClusterDescriptionChangedEvents(
110114
}
111115
}
112116

117+
/**
118+
* Waits for the cluster to be closed, which is signaled by a {@link ClusterClosedEvent}.
119+
*/
120+
public void waitForClusterClosedEvent(final Duration duration)
121+
throws InterruptedException, TimeoutException {
122+
boolean await = closedLatch.await(duration.toMillis(), TimeUnit.MILLISECONDS);
123+
if (!await) {
124+
throw new TimeoutException("Timed out waiting for cluster to close");
125+
}
126+
}
127+
113128
/**
114129
* Must be guarded by {@link #lock}.
115130
*/

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

Lines changed: 79 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,11 @@
1717
package com.mongodb.client.unified;
1818

1919
import com.mongodb.MongoNamespace;
20+
import com.mongodb.connection.ClusterDescription;
21+
import com.mongodb.connection.ServerDescription;
22+
import com.mongodb.event.ClusterClosedEvent;
23+
import com.mongodb.event.ClusterDescriptionChangedEvent;
24+
import com.mongodb.event.ClusterOpeningEvent;
2025
import com.mongodb.event.CommandEvent;
2126
import com.mongodb.event.CommandFailedEvent;
2227
import com.mongodb.event.CommandStartedEvent;
@@ -130,6 +135,47 @@ public static ContextElement ofClusterDescriptionChangedEventCount(final String
130135
return new EventCountContext("Cluster Description Changed Event Count", client, event, count);
131136
}
132137

138+
public static ContextElement ofWaitForClusterClosedEvent(final String client) {
139+
return new ContextElement() {
140+
@Override
141+
public String toString() {
142+
return "Event MatchingContext\n"
143+
+ " client: " + client + "\n"
144+
+ " expected event: ClusterClosedEvent\n";
145+
}
146+
};
147+
}
148+
149+
public static ContextElement ofTopologyEvents(final String client, final BsonArray expectedEvents,
150+
final List<?> actualEvents) {
151+
return new ContextElement() {
152+
@Override
153+
public String toString() {
154+
return "Events MatchingContext: \n"
155+
+ " client: '" + client + "'\n"
156+
+ " Expected events:\n"
157+
+ new BsonDocument("events", expectedEvents).toJson(JsonWriterSettings.builder().indent(true).build()) + "\n"
158+
+ " Actual events:\n"
159+
+ new BsonDocument("events",
160+
new BsonArray(actualEvents.stream().map(ContextElement::topologyEventToDocument).collect(Collectors.toList())))
161+
.toJson(JsonWriterSettings.builder().indent(true).build())
162+
+ "\n";
163+
}
164+
};
165+
}
166+
167+
public static ContextElement ofTopologyEvent(final BsonDocument expected, final Object actual, final int eventPosition) {
168+
return new ContextElement() {
169+
@Override
170+
public String toString() {
171+
return "Event Matching Context\n"
172+
+ " event position: " + eventPosition + "\n"
173+
+ " expected event: " + expected + "\n"
174+
+ " actual event: " + topologyEventToDocument(actual) + "\n";
175+
}
176+
};
177+
}
178+
133179
public static ContextElement ofWaitForServerMonitorEvents(final String client, final BsonDocument event, final int count) {
134180
return new EventCountContext("Wait For Server Monitor Events", client, event, count);
135181
}
@@ -152,11 +198,6 @@ public String toString() {
152198
.toJson(JsonWriterSettings.builder().indent(true).build())
153199
+ "\n";
154200
}
155-
156-
private BsonDocument toDocument(final Object event) {
157-
return new BsonDocument(EventMatcher.getEventType(event.getClass()),
158-
new BsonDocument("awaited", BsonBoolean.valueOf(EventMatcher.getAwaitedFromServerMonitorEvent(event))));
159-
}
160201
};
161202
}
162203

@@ -467,4 +508,37 @@ private static BsonDocument serverMonitorEventToDocument(final Object event) {
467508
return new BsonDocument(EventMatcher.getEventType(event.getClass()),
468509
new BsonDocument("awaited", BsonBoolean.valueOf(EventMatcher.getAwaitedFromServerMonitorEvent(event))));
469510
}
511+
512+
static BsonDocument topologyEventToDocument(final Object event) {
513+
if (event != null && !(event instanceof ClusterOpeningEvent || event instanceof ClusterDescriptionChangedEvent || event instanceof ClusterClosedEvent)) {
514+
throw new UnsupportedOperationException("Unsupported topology event: " + event.getClass().getName());
515+
}
516+
BsonDocument eventDocument = new BsonDocument();
517+
if (event instanceof ClusterDescriptionChangedEvent) {
518+
ClusterDescriptionChangedEvent changedEvent = (ClusterDescriptionChangedEvent) event;
519+
eventDocument.put("previousDescription",
520+
new BsonDocument("type", new BsonString(clusterDescriptionToString(changedEvent.getPreviousDescription()))));
521+
eventDocument.put("newDescription",
522+
new BsonDocument("type", new BsonString(clusterDescriptionToString(changedEvent.getNewDescription()))));
523+
}
524+
return new BsonDocument(EventMatcher.getEventType(event.getClass()), eventDocument);
525+
}
526+
527+
static String clusterDescriptionToString(final ClusterDescription clusterDescription) {
528+
switch (clusterDescription.getType()) {
529+
case STANDALONE:
530+
return "Single";
531+
case REPLICA_SET:
532+
return clusterDescription.getServerDescriptions().stream()
533+
.anyMatch(ServerDescription::isPrimary) ? "ReplicaSetWithPrimary" : "ReplicaSetNoPrimary";
534+
case SHARDED:
535+
return "Sharded";
536+
case LOAD_BALANCED:
537+
return "LoadBalancer";
538+
case UNKNOWN:
539+
return "Unknown";
540+
default:
541+
throw new UnsupportedOperationException("Unexpected value: " + clusterDescription.getShortDescription());
542+
}
543+
}
470544
}

0 commit comments

Comments
 (0)