GODRIVER-3035 Publish TopologyDescriptionChangedEvent on topology close.

internal/integration/unified/client_entity.go

@@ -54,6 +54,8 @@ type clientEntity struct {
 	serverHeartbeatStartedEvent []*event.ServerHeartbeatStartedEvent
 	serverHeartbeatSucceeded    []*event.ServerHeartbeatSucceededEvent
 	topologyDescriptionChanged  []*event.TopologyDescriptionChangedEvent
+	topologyOpening             []*event.TopologyOpeningEvent
+	topologyClosed              []*event.TopologyClosedEvent
 	ignoredCommands             map[string]struct{}
 	observeSensitiveCommands    *bool
 	numConnsCheckedOut          int32
@@ -161,6 +163,8 @@ func newClientEntity(ctx context.Context, em *EntityMap, entityOptions *entityOp
 			ServerHeartbeatStarted:     entity.processServerHeartbeatStartedEvent,
 			ServerHeartbeatSucceeded:   entity.processServerHeartbeatSucceededEvent,
 			TopologyDescriptionChanged: entity.processTopologyDescriptionChangedEvent,
+			TopologyOpening:            entity.processTopologyOpeningEvent,
+			TopologyClosed:             entity.processTopologyClosedEvent,
 		}
 
 		clientOpts.SetMonitor(commandMonitor).SetPoolMonitor(poolMonitor).SetServerMonitor(serverMonitor)
@@ -575,6 +579,36 @@ func (c *clientEntity) processTopologyDescriptionChangedEvent(evt *event.Topolog
 	c.addEventsCount(topologyDescriptionChangedEvent)
 }
 
+func (c *clientEntity) processTopologyOpeningEvent(evt *event.TopologyOpeningEvent) {
+	c.eventProcessMu.Lock()
+	defer c.eventProcessMu.Unlock()
+
+	if !c.getRecordEvents() {
+		return
+	}
+
+	if _, ok := c.observedEvents[topologyOpeningEvent]; ok {
+		c.topologyOpening = append(c.topologyOpening, evt)
+	}
+
+	c.addEventsCount(topologyOpeningEvent)
+}
+
+func (c *clientEntity) processTopologyClosedEvent(evt *event.TopologyClosedEvent) {
+	c.eventProcessMu.Lock()
+	defer c.eventProcessMu.Unlock()
+
+	if !c.getRecordEvents() {
+		return
+	}
+
+	if _, ok := c.observedEvents[topologyClosedEvent]; ok {
+		c.topologyClosed = append(c.topologyClosed, evt)
+	}
+
+	c.addEventsCount(topologyClosedEvent)
+}
+
 func (c *clientEntity) setRecordEvents(record bool) {
 	c.recordEvents.Store(record)
 }

internal/integration/unified/event.go

@@ -35,6 +35,8 @@ const (
 	serverHeartbeatStartedEvent     monitoringEventType = "ServerHeartbeatStartedEvent"
 	serverHeartbeatSucceededEvent   monitoringEventType = "ServerHeartbeatSucceededEvent"
 	topologyDescriptionChangedEvent monitoringEventType = "TopologyDescriptionChangedEvent"
+	topologyOpeningEvent            monitoringEventType = "TopologyOpeningEvent"
+	topologyClosedEvent             monitoringEventType = "TopologyClosedEvent"
 )
 
 func monitoringEventTypeFromString(eventStr string) (monitoringEventType, bool) {
@@ -77,6 +79,10 @@ func monitoringEventTypeFromString(eventStr string) (monitoringEventType, bool)
 		return serverHeartbeatSucceededEvent, true
 	case "topologydescriptionchangedevent":
 		return topologyDescriptionChangedEvent, true
+	case "topologyopeningevent":
+		return topologyOpeningEvent, true
+	case "topologyclosedevent":
+		return topologyClosedEvent, true
 	default:
 		return "", false
 	}

internal/integration/unified/event_verification.go

@@ -87,7 +87,16 @@ type sdamEvent struct {
 		Awaited *bool `bson:"awaited"`
 	} `bson:"serverHeartbeatFailedEvent"`
 
-	TopologyDescriptionChangedEvent *struct{} `bson:"topologyDescriptionChangedEvent"`
+	TopologyDescriptionChangedEvent *struct {
+		PreviousDescription *struct {
+			Type *string `bson:"type"`
+		} `bson:"previousDescription"`
+		NewDescription *struct {
+			Type *string `bson:"type"`
+		} `bson:"newDescription"`
+	} `bson:"topologyDescriptionChangedEvent"`
+	TopologyOpeningEvent *struct{} `bson:"topologyOpeningEvent"`
+	TopologyClosedEvent  *struct{} `bson:"topologyClosedEvent"`
 }
 
 type expectedEvents struct {
@@ -468,7 +477,7 @@ func getNextServerHeartbeatSucceededEvent(
 		return nil, nil, errors.New("no heartbeat succeeded event published")
 	}
 
-	return events[0], events[:1], nil
+	return events[0], events[1:], nil
 }
 
 func getNextServerHeartbeatFailedEvent(
@@ -478,7 +487,7 @@ func getNextServerHeartbeatFailedEvent(
 		return nil, nil, errors.New("no heartbeat failed event published")
 	}
 
-	return events[0], events[:1], nil
+	return events[0], events[1:], nil
 }
 
 func getNextTopologyDescriptionChangedEvent(
@@ -488,7 +497,27 @@ func getNextTopologyDescriptionChangedEvent(
 		return nil, nil, errors.New("no topology description changed event published")
 	}
 
-	return events[0], events[:1], nil
+	return events[0], events[1:], nil
+}
+
+func getNextTopologyOpeningEvent(
+	events []*event.TopologyOpeningEvent,
+) (*event.TopologyOpeningEvent, []*event.TopologyOpeningEvent, error) {
+	if len(events) == 0 {
+		return nil, nil, errors.New("no topology opening event published")
+	}
+
+	return events[0], events[1:], nil
+}
+
+func getNextTopologyClosedEvent(
+	events []*event.TopologyClosedEvent,
+) (*event.TopologyClosedEvent, []*event.TopologyClosedEvent, error) {
+	if len(events) == 0 {
+		return nil, nil, errors.New("no topology closed event published")
+	}
+
+	return events[0], events[1:], nil
 }
 
 func verifySDAMEvents(client *clientEntity, expectedEvents *expectedEvents) error {
@@ -498,9 +527,21 @@ func verifySDAMEvents(client *clientEntity, expectedEvents *expectedEvents) erro
 		succeeded = client.serverHeartbeatSucceeded
 		failed    = client.serverHeartbeatFailedEvent
 		tchanged  = client.topologyDescriptionChanged
+		topening  = client.topologyOpening
+		tclosed   = client.topologyClosed
 	)
 
-	vol := func() int { return len(changed) + len(started) + len(succeeded) + len(failed) + len(tchanged) }
+	vol := func() int {
+		var count int
+		count += len(changed)
+		count += len(started)
+		count += len(succeeded)
+		count += len(failed)
+		count += len(tchanged)
+		count += len(topening)
+		count += len(tclosed)
+		return count
+	}
 
 	if len(expectedEvents.SDAMEvents) == 0 && vol() != 0 {
 		return fmt.Errorf("expected no sdam events to be sent but got %s", stringifyEventsForClient(client))
@@ -569,7 +610,23 @@ func verifySDAMEvents(client *clientEntity, expectedEvents *expectedEvents) erro
 				return newEventVerificationError(idx, client, "want awaited %v, got %v", *want, got.Awaited)
 			}
 		case evt.TopologyDescriptionChangedEvent != nil:
-			if _, tchanged, err = getNextTopologyDescriptionChangedEvent(tchanged); err != nil {
+			var got *event.TopologyDescriptionChangedEvent
+			if got, tchanged, err = getNextTopologyDescriptionChangedEvent(tchanged); err != nil {
+				return newEventVerificationError(idx, client, "failed to get next description changed event: %v", err.Error())
+			}
+
+			if want := evt.TopologyDescriptionChangedEvent.PreviousDescription; want != nil && want.Type != nil && *want.Type != got.PreviousDescription.Kind {
+				return newEventVerificationError(idx, client, "want previous description %v, got %v", *want.Type, got.PreviousDescription.Kind)
+			}
+			if want := evt.TopologyDescriptionChangedEvent.NewDescription; want != nil && want.Type != nil && *want.Type != got.NewDescription.Kind {
+				return newEventVerificationError(idx, client, "want new description %v, got %v", *want.Type, got.NewDescription.Kind)
+			}
+		case evt.TopologyOpeningEvent != nil:
+			if _, topening, err = getNextTopologyOpeningEvent(topening); err != nil {
+				return newEventVerificationError(idx, client, "failed to get next description changed event: %v", err.Error())
+			}
+		case evt.TopologyClosedEvent != nil:
+			if _, tclosed, err = getNextTopologyClosedEvent(tclosed); err != nil {
 				return newEventVerificationError(idx, client, "failed to get next description changed event: %v", err.Error())
 			}
 		}

testdata/server-discovery-and-monitoring/unified/loadbalanced-emit-topology-changed-before-close.json

@@ -0,0 +1,88 @@
+{
+  "description": "loadbalanced-emit-topology-description-changed-before-close",
+  "schemaVersion": "1.20",
+  "runOnRequirements": [
+    {
+      "topologies": [
+        "load-balanced"
+      ],
+      "minServerVersion": "4.4"
+    }
+  ],
+  "tests": [
+    {
+      "description": "Topology lifecycle",
+      "operations": [
+        {
+          "name": "createEntities",
+          "object": "testRunner",
+          "arguments": {
+            "entities": [
+              {
+                "client": {
+                  "id": "client",
+                  "observeEvents": [
+                    "topologyDescriptionChangedEvent",
+                    "topologyOpeningEvent",
+                    "topologyClosedEvent"
+                  ]
+                }
+              }
+            ]
+          }
+        },
+        {
+          "name": "waitForEvent",
+          "object": "testRunner",
+          "arguments": {
+            "client": "client",
+            "event": {
+              "topologyDescriptionChangedEvent": {}
+            },
+            "count": 2
+          }
+        },
+        {
+          "name": "close",
+          "object": "client"
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "eventType": "sdam",
+          "events": [
+            {
+              "topologyOpeningEvent": {}
+            },
+            {
+              "topologyDescriptionChangedEvent": {
+                "previousDescription": {
+                  "type": "Unknown"
+                },
+                "newDescription": {}
+              }
+            },
+            {
+              "topologyDescriptionChangedEvent": {
+                "newDescription": {
+                  "type": "LoadBalanced"
+                }
+              }
+            },
+            {
+              "topologyDescriptionChangedEvent": {
+                "newDescription": {
+                  "type": "Unknown"
+                }
+              }
+            },
+            {
+              "topologyClosedEvent": {}
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

testdata/server-discovery-and-monitoring/unified/loadbalanced-emit-topology-changed-before-close.yml

@@ -0,0 +1,49 @@
+description: "loadbalanced-emit-topology-description-changed-before-close"
+
+schemaVersion: "1.20"
+
+runOnRequirements:
+  - topologies:
+      - load-balanced
+    minServerVersion: "4.4" # awaitable hello
+
+tests:
+  - description: "Topology lifecycle"
+    operations:
+      - name: createEntities
+        object: testRunner
+        arguments:
+          entities:
+            - client:
+                id: &client client
+                observeEvents:
+                  - topologyDescriptionChangedEvent
+                  - topologyOpeningEvent
+                  - topologyClosedEvent
+      # ensure the topology has been fully discovered before closing the client.
+      # expected events are initial server discovery and server connect event.
+      - name: waitForEvent
+        object: testRunner
+        arguments:
+          client: *client
+          event:
+            topologyDescriptionChangedEvent: {}
+          count: 2
+      - name: close
+        object: *client
+    expectEvents:
+      - client: *client
+        eventType: sdam
+        events:
+          - topologyOpeningEvent: {}
+          - topologyDescriptionChangedEvent: # unknown -> unknown w disconnected server
+              previousDescription: 
+                type: "Unknown"
+              newDescription: {}
+          - topologyDescriptionChangedEvent:  # unknown w disconnected server -> loadBalanced
+              newDescription:
+                type: "LoadBalanced"
+          - topologyDescriptionChangedEvent:  # loadbalanced -> unknown
+              newDescription:
+                type: "Unknown"
+          - topologyClosedEvent: {}

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

@@ -132,6 +132,22 @@
                 }
               }
             },
+            {
+              "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-loadbalanced.yml

@@ -67,6 +67,13 @@ tests:
               topologyId: { $$exists: true }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
+          - level: debug
+            component: topology
+            data:
+              message: "Topology description changed"
+              topologyId: { $$exists: true }
+              previousDescription: { $$exists: true } # loadBalanced topology
+              newDescription: { $$exists: true } # unknown topology
           - level: debug
             component: topology
             data:

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

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