RUST-645 / RUST-1245 Add Atlas planned maintenance testing workload executor (#657)

Author: isabelatkinson

Cargo.toml

@@ -23,13 +23,24 @@ exclude = [
     ".evergreen/**",
     ".gitignore",
     "src/test/**",
-    "tests/**"
+    "tests/**",
 ]
 
 [features]
 default = ["tokio-runtime"]
-tokio-runtime = ["tokio/macros", "tokio/net", "tokio/rt", "tokio/time", "serde_bytes"]
-async-std-runtime = ["async-std", "async-std/attributes", "async-std-resolver", "tokio-util/compat"]
+tokio-runtime = [
+    "tokio/macros",
+    "tokio/net",
+    "tokio/rt",
+    "tokio/time",
+    "serde_bytes",
+]
+async-std-runtime = [
+    "async-std",
+    "async-std/attributes",
+    "async-std-resolver",
+    "tokio-util/compat",
+]
 sync = ["async-std-runtime"]
 tokio-sync = ["tokio-runtime"]
 openssl-tls = ["openssl", "openssl-probe", "tokio-openssl"]
@@ -76,7 +87,7 @@ rustls-pemfile = "0.3.0"
 serde_with = "1.3.1"
 sha-1 = "0.10.0"
 sha2 = "0.10.2"
-snap = { version = "1.0.5", optional = true}
+snap = { version = "1.0.5", optional = true }
 socket2 = "0.4.0"
 stringprep = "0.1.2"
 strsim = "0.10.0"
@@ -138,13 +149,15 @@ features = ["v4"]
 [dev-dependencies]
 approx = "0.5.1"
 async_once = "0.2.6"
+ctrlc = "3.2.2"
 derive_more = "0.99.13"
 function_name = "0.2.1"
 futures = "0.3"
 home = "0.5"
 pretty_assertions = "1.1.0"
 serde_json = "1.0.64"
 semver = "1.0.0"
+time = "0.3.9"
 
 [package.metadata.docs.rs]
 rustdoc-args = ["--cfg", "docsrs"]

src/bson_util/mod.rs

@@ -9,7 +9,7 @@ use serde::{de::Error as SerdeDeError, ser, Deserialize, Deserializer, Serialize
 
 use crate::{
     bson::{doc, Bson, Document},
-    error::{ErrorKind, Result},
+    error::{Error, ErrorKind, Result},
     runtime::SyncLittleEndianRead,
 };
 
@@ -217,6 +217,14 @@ pub(crate) fn read_document_bytes<R: Read>(mut reader: R) -> Result<Vec<u8>> {
     Ok(bytes)
 }
 
+/// Serializes an Error as a string.
+pub(crate) fn serialize_error_as_string<S: Serializer>(
+    val: &Error,
+    serializer: S,
+) -> std::result::Result<S::Ok, S::Error> {
+    serializer.serialize_str(&val.to_string())
+}
+
 #[cfg(test)]
 mod test {
     use crate::bson_util::num_decimal_digits;

src/cmap/conn/mod.rs

@@ -8,6 +8,7 @@ use std::{
 };
 
 use derivative::Derivative;
+use serde::Serialize;
 use tokio::sync::{mpsc, Mutex};
 
 use self::wire::Message;
@@ -37,7 +38,7 @@ pub(crate) use stream_description::StreamDescription;
 pub(crate) use wire::next_request_id;
 
 /// User-facing information about a connection to the database.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize)]
 pub struct ConnectionInfo {
     /// A driver-generated identifier that uniquely identifies the connection.
     pub id: u32,

src/cmap/test/event.rs

@@ -3,7 +3,7 @@ use std::{
     time::Duration,
 };
 
-use serde::{de::Unexpected, Deserialize, Deserializer};
+use serde::{de::Unexpected, Deserialize, Deserializer, Serialize};
 
 use crate::{event::cmap::*, options::ServerAddress, runtime};
 use tokio::sync::broadcast::error::{RecvError, SendError};
@@ -154,6 +154,27 @@ pub enum Event {
     ConnectionCheckedIn(ConnectionCheckedInEvent),
 }
 
+impl Serialize for Event {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        match self {
+            Self::PoolCreated(event) => event.serialize(serializer),
+            Self::PoolClosed(event) => event.serialize(serializer),
+            Self::PoolReady(event) => event.serialize(serializer),
+            Self::ConnectionCreated(event) => event.serialize(serializer),
+            Self::ConnectionReady(event) => event.serialize(serializer),
+            Self::ConnectionClosed(event) => event.serialize(serializer),
+            Self::ConnectionCheckOutStarted(event) => event.serialize(serializer),
+            Self::ConnectionCheckOutFailed(event) => event.serialize(serializer),
+            Self::ConnectionCheckedOut(event) => event.serialize(serializer),
+            Self::PoolCleared(event) => event.serialize(serializer),
+            Self::ConnectionCheckedIn(event) => event.serialize(serializer),
+        }
+    }
+}
+
 impl Event {
     pub fn name(&self) -> &'static str {
         match self {
@@ -170,6 +191,24 @@ impl Event {
             Event::ConnectionCheckedIn(_) => "ConnectionCheckedIn",
         }
     }
+
+    // The names in drivers-atlas-testing tests are slightly different than those used in spec
+    // tests.
+    pub fn planned_maintenance_testing_name(&self) -> &'static str {
+        match self {
+            Event::PoolCreated(_) => "PoolCreatedEvent",
+            Event::PoolReady(_) => "PoolReadyEvent",
+            Event::PoolCleared(_) => "PoolClearedEvent",
+            Event::PoolClosed(_) => "PoolClosedEvent",
+            Event::ConnectionCreated(_) => "ConnectionCreatedEvent",
+            Event::ConnectionReady(_) => "ConnectionReadyEvent",
+            Event::ConnectionClosed(_) => "ConnectionClosedEvent",
+            Event::ConnectionCheckOutStarted(_) => "ConnectionCheckOutStartedEvent",
+            Event::ConnectionCheckOutFailed(_) => "ConnectionCheckOutFailedEvent",
+            Event::ConnectionCheckedOut(_) => "ConnectionCheckedOutEvent",
+            Event::ConnectionCheckedIn(_) => "ConnectionCheckedInEvent",
+        }
+    }
 }
 
 #[derive(Debug, Deserialize)]

src/event/cmap.rs

@@ -3,7 +3,7 @@
 
 use std::time::Duration;
 
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
 use crate::{bson::oid::ObjectId, bson_util, options::ServerAddress};
 
@@ -20,12 +20,12 @@ fn empty_address() -> ServerAddress {
 }
 
 /// Event emitted when a connection pool is created.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[non_exhaustive]
 pub struct PoolCreatedEvent {
     /// The address of the server that the pool's connections will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The options used for the pool.
@@ -34,7 +34,7 @@ pub struct PoolCreatedEvent {
 
 /// Contains the options for creating a connection pool. While these options are specified at the
 /// client-level, `ConnectionPoolOptions` is exposed for the purpose of CMAP event handling.
-#[derive(Clone, Default, Deserialize, Debug, PartialEq)]
+#[derive(Clone, Default, Deserialize, Debug, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub struct ConnectionPoolOptions {
@@ -62,46 +62,46 @@ pub struct ConnectionPoolOptions {
 }
 
 /// Event emitted when a connection pool becomes ready.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[non_exhaustive]
 pub struct PoolReadyEvent {
     /// The address of the server that the pool's connections will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 }
 
 /// Event emitted when a connection pool is cleared.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[non_exhaustive]
 pub struct PoolClearedEvent {
     /// The address of the server that the pool's connections will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// If the connection is to a load balancer, the id of the selected backend.
     pub service_id: Option<ObjectId>,
 }
 
 /// Event emitted when a connection pool is cleared.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[non_exhaustive]
 pub struct PoolClosedEvent {
     /// The address of the server that the pool's connections will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 }
 
 /// Event emitted when a connection is created.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub struct ConnectionCreatedEvent {
     /// The address of the server that the connection will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The unique ID of the connection. This is not used for anything internally, but can be used
@@ -112,13 +112,13 @@ pub struct ConnectionCreatedEvent {
 
 /// Event emitted when a connection is ready to be used. This indicates that all the necessary
 /// prerequisites for using a connection (handshake, authentication, etc.) have been completed.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub struct ConnectionReadyEvent {
     /// The address of the server that the connection is connected to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The unique ID of the connection. This is not used for anything internally, but can be used
@@ -128,13 +128,13 @@ pub struct ConnectionReadyEvent {
 }
 
 /// Event emitted when a connection is closed.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub struct ConnectionClosedEvent {
     /// The address of the server that the connection was connected to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The unique ID of the connection. This is not used for anything internally, but can be used
@@ -147,7 +147,7 @@ pub struct ConnectionClosedEvent {
 }
 
 /// The reasons that a connection may be closed.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub enum ConnectionClosedReason {
@@ -168,30 +168,30 @@ pub enum ConnectionClosedReason {
 }
 
 /// Event emitted when a thread begins checking out a connection to use for an operation.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[non_exhaustive]
 pub struct ConnectionCheckoutStartedEvent {
     /// The address of the server that the connection will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 }
 
 /// Event emitted when a thread is unable to check out a connection.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[non_exhaustive]
 pub struct ConnectionCheckoutFailedEvent {
     /// The address of the server that the connection would have connected to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The reason a connection was unable to be checked out.
     pub reason: ConnectionCheckoutFailedReason,
 }
 
 /// The reasons a connection may not be able to be checked out.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub enum ConnectionCheckoutFailedReason {
@@ -204,13 +204,13 @@ pub enum ConnectionCheckoutFailedReason {
 }
 
 /// Event emitted when a connection is successfully checked out.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub struct ConnectionCheckedOutEvent {
     /// The address of the server that the connection will connect to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The unique ID of the connection. This is not used for anything internally, but can be used
@@ -220,13 +220,13 @@ pub struct ConnectionCheckedOutEvent {
 }
 
 /// Event emitted when a connection is checked back into a connection pool.
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 #[non_exhaustive]
 pub struct ConnectionCheckedInEvent {
     /// The address of the server that the connection was connected to.
     #[serde(default = "self::empty_address")]
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     pub address: ServerAddress,
 
     /// The unique ID of the connection. This is not used for anything internally, but can be used