RUST-910 Expose server connectionId in command monitoring events (#682)

Author: kmahar

src/cmap/conn/mod.rs

@@ -39,10 +39,16 @@ pub(crate) use wire::next_request_id;
 
 /// User-facing information about a connection to the database.
 #[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+#[non_exhaustive]
 pub struct ConnectionInfo {
     /// A driver-generated identifier that uniquely identifies the connection.
     pub id: u32,
 
+    /// A server-generated identifier that uniquely identifies the connection. Available on server
+    /// versions 4.2+. This may be used to correlate driver connections with server logs.
+    pub server_id: Option<i32>,
+
     /// The address that the connection is connected to.
     pub address: ServerAddress,
 }
@@ -51,7 +57,11 @@ pub struct ConnectionInfo {
 #[derive(Derivative)]
 #[derivative(Debug)]
 pub(crate) struct Connection {
+    /// Driver-generated ID for the connection.
     pub(super) id: u32,
+    /// Server-generated ID for the connection.
+    pub(crate) server_id: Option<i32>,
+
     pub(crate) address: ServerAddress,
     pub(crate) generation: ConnectionGeneration,
 
@@ -109,6 +119,7 @@ impl Connection {
 
         let conn = Self {
             id,
+            server_id: None,
             generation: ConnectionGeneration::Normal(generation),
             pool_manager: None,
             command_executing: false,
@@ -174,6 +185,7 @@ impl Connection {
     pub(crate) fn info(&self) -> ConnectionInfo {
         ConnectionInfo {
             id: self.id,
+            server_id: self.server_id,
             address: self.address.clone(),
         }
     }
@@ -370,6 +382,7 @@ impl Connection {
     fn take(&mut self) -> Connection {
         Connection {
             id: self.id,
+            server_id: self.server_id,
             address: self.address.clone(),
             generation: self.generation.clone(),
             stream: std::mem::replace(&mut self.stream, AsyncStream::Null),

src/cmap/establish/handshake/mod.rs

@@ -269,6 +269,8 @@ impl Handshaker {
             }
         }
 
+        conn.server_id = hello_reply.command_response.connection_id;
+
         Ok(HandshakeResult {
             hello_reply,
             first_round,

src/event/command.rs

@@ -8,10 +8,11 @@ use serde::Serialize;
 use crate::{
     bson::{oid::ObjectId, Document},
     bson_util::serialize_error_as_string,
-    cmap::ConnectionInfo,
     error::Error,
 };
 
+pub use crate::cmap::ConnectionInfo;
+
 /// An event that triggers when a database command is initiated.
 #[derive(Clone, Debug, Serialize)]
 #[serde(rename_all = "camelCase")]

src/hello.rs

@@ -237,6 +237,10 @@ pub(crate) struct HelloCommandResponse {
 
     /// The maximum permitted size of a BSON wire protocol message.
     pub max_message_size_bytes: i32,
+
+    /// The server-generated ID for the connection the "hello" command was run on.
+    /// Present on server versions 4.2+.
+    pub connection_id: Option<i32>,
 }
 
 impl PartialEq for HelloCommandResponse {

src/sdam/description/topology/test/sdam.rs

@@ -126,6 +126,7 @@ impl From<TestHelloCommandResponse> for HelloCommandResponse {
             compressors: None,
             hello_ok: test.hello_ok,
             max_message_size_bytes: 48 * 1024 * 1024,
+            connection_id: None,
         }
     }
 }

src/test/spec/json/command-monitoring/unified/pre-42-server-connection-id.json

@@ -0,0 +1,101 @@
+{
+  "description": "pre-42-server-connection-id",
+  "schemaVersion": "1.6",
+  "runOnRequirements": [
+    {
+      "maxServerVersion": "4.0.99"
+    }
+  ],
+  "createEntities": [
+    {
+      "client": {
+        "id": "client",
+        "observeEvents": [
+          "commandStartedEvent",
+          "commandSucceededEvent",
+          "commandFailedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "database",
+        "client": "client",
+        "databaseName": "server-connection-id-tests"
+      }
+    },
+    {
+      "collection": {
+        "id": "collection",
+        "database": "database",
+        "collectionName": "coll"
+      }
+    }
+  ],
+  "initialData": [
+    {
+      "databaseName": "server-connection-id-tests",
+      "collectionName": "coll",
+      "documents": []
+    }
+  ],
+  "tests": [
+    {
+      "description": "command events do not include server connection id",
+      "operations": [
+        {
+          "name": "insertOne",
+          "object": "collection",
+          "arguments": {
+            "document": {
+              "x": 1
+            }
+          }
+        },
+        {
+          "name": "find",
+          "object": "collection",
+          "arguments": {
+            "filter": {
+              "$or": true
+            }
+          },
+          "expectError": {
+            "isError": true
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "insert",
+                "hasServerConnectionId": false
+              }
+            },
+            {
+              "commandSucceededEvent": {
+                "commandName": "insert",
+                "hasServerConnectionId": false
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "hasServerConnectionId": false
+              }
+            },
+            {
+              "commandFailedEvent": {
+                "commandName": "find",
+                "hasServerConnectionId": false
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

src/test/spec/json/command-monitoring/unified/pre-42-server-connection-id.yml

@@ -0,0 +1,56 @@
+description: "pre-42-server-connection-id"
+
+schemaVersion: "1.6"
+
+runOnRequirements:
+  - maxServerVersion: "4.0.99"
+
+createEntities:
+  - client:
+      id: &client client
+      observeEvents:
+        - commandStartedEvent
+        - commandSucceededEvent
+        - commandFailedEvent
+  - database:
+      id: &database database
+      client: *client
+      databaseName: &databaseName server-connection-id-tests
+  - collection:
+      id: &collection collection
+      database: *database
+      collectionName: &collectionName coll
+
+initialData:
+  - databaseName: *databaseName
+    collectionName: *collectionName
+    documents: []
+
+tests:
+  - description: "command events do not include server connection id"
+    operations:
+      - name: insertOne
+        object: *collection
+        arguments:
+            document: { x: 1 }
+      - name: find
+        object: *collection
+        arguments:
+          filter: { $or: true }
+        expectError:
+          isError: true
+    expectEvents:
+      - client: *client
+        events:
+          - commandStartedEvent:
+              commandName: insert
+              hasServerConnectionId: false
+          - commandSucceededEvent:
+              commandName: insert
+              hasServerConnectionId: false
+          - commandStartedEvent:
+              commandName: find
+              hasServerConnectionId: false
+          - commandFailedEvent:
+              commandName: find
+              hasServerConnectionId: false

src/test/spec/json/command-monitoring/unified/server-connection-id.json

@@ -0,0 +1,101 @@
+{
+  "description": "server-connection-id",
+  "schemaVersion": "1.6",
+  "runOnRequirements": [
+    {
+      "minServerVersion": "4.2"
+    }
+  ],
+  "createEntities": [
+    {
+      "client": {
+        "id": "client",
+        "observeEvents": [
+          "commandStartedEvent",
+          "commandSucceededEvent",
+          "commandFailedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "database",
+        "client": "client",
+        "databaseName": "server-connection-id-tests"
+      }
+    },
+    {
+      "collection": {
+        "id": "collection",
+        "database": "database",
+        "collectionName": "coll"
+      }
+    }
+  ],
+  "initialData": [
+    {
+      "databaseName": "server-connection-id-tests",
+      "collectionName": "coll",
+      "documents": []
+    }
+  ],
+  "tests": [
+    {
+      "description": "command events include server connection id",
+      "operations": [
+        {
+          "name": "insertOne",
+          "object": "collection",
+          "arguments": {
+            "document": {
+              "x": 1
+            }
+          }
+        },
+        {
+          "name": "find",
+          "object": "collection",
+          "arguments": {
+            "filter": {
+              "$or": true
+            }
+          },
+          "expectError": {
+            "isError": true
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "insert",
+                "hasServerConnectionId": true
+              }
+            },
+            {
+              "commandSucceededEvent": {
+                "commandName": "insert",
+                "hasServerConnectionId": true
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "hasServerConnectionId": true
+              }
+            },
+            {
+              "commandFailedEvent": {
+                "commandName": "find",
+                "hasServerConnectionId": true
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}