RUST-580 Add tracing integration and command tracing events (#757)

Author: kmahar

.evergreen/compile-only.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 
 set -o errexit
+set -o xtrace
 
 source ./.evergreen/configure-rust.sh
 rustup update $RUST_VERSION
@@ -10,21 +11,9 @@ if [  "$MSRV" = "true" ]; then
     cp .evergreen/MSRV-Cargo.lock Cargo.lock
 fi
 
-if [ "$ASYNC_RUNTIME" = "tokio" ]; then
-    FEATURE_FLAGS=snappy-compression,zlib-compression
+source ./.evergreen/feature-combinations.sh
 
-    # Zstd requires Rust version 1.54
-    if [[ $RUST_VERSION == "nightly" ]]; then
-        FEATURE_FLAGS=$FEATURE_FLAGS,zstd-compression
-    fi
-
-    rustup run $RUST_VERSION cargo build --features $FEATURE_FLAGS
-    rustup run $RUST_VERSION cargo build --features tokio-sync,$FEATURE_FLAGS
-
-elif [ "$ASYNC_RUNTIME" = "async-std" ]; then
-    rustup run $RUST_VERSION cargo build --no-default-features --features async-std-runtime
-    rustup run $RUST_VERSION cargo build --no-default-features --features sync
-else
-    echo "invalid async runtime: ${ASYNC_RUNTIME}" >&2
-    exit 1
-fi
+# build with all available features to ensure all optional dependencies are brought in too.
+for ((i = 0; i < ${#FEATURE_COMBINATIONS[@]}; i++)); do
+    rustup run $RUST_VERSION cargo build ${FEATURE_COMBINATIONS[$i]}
+done

.evergreen/config.yml

@@ -548,7 +548,7 @@ functions:
         working_dir: "src"
         script: |
           ${PREPARE_SHELL}
-          ASYNC_RUNTIME=${ASYNC_RUNTIME} RUST_VERSION=${RUST_VERSION} MSRV=${MSRV} .evergreen/compile-only.sh
+          RUST_VERSION=${RUST_VERSION} MSRV=${MSRV} .evergreen/compile-only.sh
 
   "check cargo deny":
     - command: shell.exec
@@ -1858,9 +1858,8 @@ buildvariants:
   matrix_spec:
     os:
       - ubuntu-18.04
-    async-runtime: "*"
     extra-rust-versions: "*"
-  display_name: "! Compile on Rust ${extra-rust-versions} with ${async-runtime}"
+  display_name: "! Compile on Rust ${extra-rust-versions}"
   tasks:
     - "compile-only"
 

.evergreen/feature-combinations.sh

@@ -6,5 +6,5 @@
 export FEATURE_COMBINATIONS=(
     '' # default features
     '--no-default-features --features async-std-runtime,sync' # features that conflict w/ default features
-    '--features tokio-sync,zstd-compression,snappy-compression,zlib-compression,openssl-tls,aws-auth,csfle' # additive features
+    '--features tokio-sync,zstd-compression,snappy-compression,zlib-compression,openssl-tls,aws-auth,tracing-unstable,csfle' # additive features
 )

.evergreen/run-tests.sh

@@ -10,7 +10,7 @@ if [ "$SINGLE_THREAD" = true ]; then
 	OPTIONS="$OPTIONS --test-threads=1"
 fi
 
-FEATURE_FLAGS="zstd-compression,snappy-compression,zlib-compression,${TLS_FEATURE}"
+FEATURE_FLAGS="zstd-compression,snappy-compression,zlib-compression,tracing-unstable,${TLS_FEATURE}"
 
 if [ "$ASYNC_RUNTIME" = "tokio" ]; then
     ASYNC_FEATURE_FLAGS=${FEATURE_FLAGS}

Cargo.toml

@@ -70,6 +70,11 @@ snappy-compression = ["snap"]
 # DO NOT USE; see https://jira.mongodb.org/browse/RUST-569 for the status of CSFLE support in the Rust driver.
 csfle = ["mongocrypt", "rayon", "num_cpus"]
 
+# DO NOT USE; see https://jira.mongodb.org/browse/RUST-580 for the status of tracing/logging support in the Rust driver.
+# Enables support for emitting tracing events.
+# TODO: pending https://github.com/tokio-rs/tracing/issues/2036 stop depending directly on log.
+tracing-unstable = ["tracing", "log"]
+
 [dependencies]
 async-trait = "0.1.42"
 base64 = "0.13.0"
@@ -85,6 +90,7 @@ futures-executor = "0.3.14"
 hex = "0.4.0"
 hmac = "0.12.1"
 lazy_static = "1.4.0"
+log = { version = "0.4.17", optional = true }
 md-5 = "0.10.1"
 mongocrypt = { git = "https://github.com/mongodb/libmongocrypt-rust.git", branch = "main", optional = true }
 num_cpus = { version = "1.13.1", optional = true }
@@ -105,6 +111,7 @@ strsim = "0.10.0"
 take_mut = "0.2.2"
 thiserror = "1.0.24"
 tokio-openssl = { version = "0.6.3", optional = true }
+tracing = { version = "0.1.36", optional = true }
 trust-dns-proto = "0.21.2"
 trust-dns-resolver = "0.21.2"
 typed-builder = "0.10.0"
@@ -172,6 +179,7 @@ serde = { version = "*", features = ["rc"] }
 serde_json = "1.0.64"
 semver = "1.0.0"
 time = "0.3.9"
+tracing-subscriber = "0.3.16"
 regex = "1.6.0"
 
 [package.metadata.docs.rs]

src/client/executor.rs

@@ -34,7 +34,12 @@ use crate::{
         TRANSIENT_TRANSACTION_ERROR,
         UNKNOWN_TRANSACTION_COMMIT_RESULT,
     },
-    event::command::{CommandFailedEvent, CommandStartedEvent, CommandSucceededEvent},
+    event::command::{
+        CommandEvent,
+        CommandFailedEvent,
+        CommandStartedEvent,
+        CommandSucceededEvent,
+    },
     hello::LEGACY_HELLO_COMMAND_NAME_LOWERCASE,
     operation::{
         AbortTransaction,
@@ -624,23 +629,21 @@ impl Client {
             bytes: serialized,
         };
 
-        self.emit_command_event(|handler| {
+        self.emit_command_event(|| {
             let command_body = if should_redact {
                 Document::new()
             } else {
                 Document::from_reader(raw_cmd.bytes.as_slice())
                     .unwrap_or_else(|e| doc! { "serialization error": e.to_string() })
             };
-            let command_started_event = CommandStartedEvent {
+            CommandEvent::Started(CommandStartedEvent {
                 command: command_body,
                 db: raw_cmd.target_db.clone(),
                 command_name: raw_cmd.name.clone(),
                 request_id,
                 connection: connection_info.clone(),
                 service_id,
-            };
-
-            handler.handle_command_started_event(command_started_event);
+            })
         });
 
         let start_time = Instant::now();
@@ -727,17 +730,20 @@ impl Client {
 
         match command_result {
             Err(mut err) => {
-                self.emit_command_event(|handler| {
-                    let command_failed_event = CommandFailedEvent {
+                self.emit_command_event(|| {
+                    let mut err = err.clone();
+                    if should_redact {
+                        err.redact();
+                    }
+
+                    CommandEvent::Failed(CommandFailedEvent {
                         duration,
-                        command_name: cmd_name,
-                        failure: err.clone(),
+                        command_name: cmd_name.clone(),
+                        failure: err,
                         request_id,
-                        connection: connection_info,
+                        connection: connection_info.clone(),
                         service_id,
-                    };
-
-                    handler.handle_command_failed_event(command_failed_event);
+                    })
                 });
 
                 if let Some(ref mut session) = session {
@@ -750,7 +756,7 @@ impl Client {
                 op.handle_error(err)
             }
             Ok(response) => {
-                self.emit_command_event(|handler| {
+                self.emit_command_event(|| {
                     let reply = if should_redact {
                         Document::new()
                     } else {
@@ -759,15 +765,14 @@ impl Client {
                             .unwrap_or_else(|e| doc! { "deserialization error": e.to_string() })
                     };
 
-                    let command_succeeded_event = CommandSucceededEvent {
+                    CommandEvent::Succeeded(CommandSucceededEvent {
                         duration,
                         reply,
                         command_name: cmd_name.clone(),
                         request_id,
-                        connection: connection_info,
+                        connection: connection_info.clone(),
                         service_id,
-                    };
-                    handler.handle_command_succeeded_event(command_succeeded_event);
+                    })
                 });
 
                 #[cfg(feature = "csfle")]

src/client/mod.rs

@@ -14,6 +14,13 @@ use derivative::Derivative;
 
 #[cfg(test)]
 use crate::options::ServerAddress;
+#[cfg(feature = "tracing-unstable")]
+use crate::trace::{
+    command::CommandTracingEventEmitter,
+    trace_or_log_enabled,
+    TracingOrLogLevel,
+    COMMAND_TRACING_EVENT_TARGET,
+};
 use crate::{
     bson::Document,
     change_stream::{
@@ -25,7 +32,7 @@ use crate::{
     concern::{ReadConcern, WriteConcern},
     db::Database,
     error::{ErrorKind, Result},
-    event::command::CommandEventHandler,
+    event::command::{handle_command_event, CommandEvent},
     operation::{AggregateTarget, ListDatabases},
     options::{
         ClientOptions,
@@ -39,6 +46,7 @@ use crate::{
     sdam::{server_selection, SelectedServer, SessionSupportStatus, Topology},
     ClientSession,
 };
+
 pub(crate) use executor::{HELLO_COMMAND_NAMES, REDACTED_COMMANDS};
 pub(crate) use session::{ClusterTime, SESSIONS_UNSUPPORTED_COMMANDS};
 
@@ -145,9 +153,42 @@ impl Client {
         Ok(client)
     }
 
-    pub(crate) fn emit_command_event(&self, emit: impl FnOnce(&Arc<dyn CommandEventHandler>)) {
+    #[cfg(not(feature = "tracing-unstable"))]
+    pub(crate) fn emit_command_event(&self, generate_event: impl FnOnce() -> CommandEvent) {
         if let Some(ref handler) = self.inner.options.command_event_handler {
-            emit(handler);
+            let event = generate_event();
+            handle_command_event(handler.as_ref(), event);
+        }
+    }
+
+    #[cfg(feature = "tracing-unstable")]
+    pub(crate) fn emit_command_event(&self, generate_event: impl FnOnce() -> CommandEvent) {
+        let tracing_emitter = if trace_or_log_enabled!(
+            target: COMMAND_TRACING_EVENT_TARGET,
+            TracingOrLogLevel::Debug
+        ) {
+            Some(CommandTracingEventEmitter::new(
+                self.inner.options.tracing_max_document_length_bytes,
+                self.inner.topology.id,
+            ))
+        } else {
+            None
+        };
+        let apm_event_handler = self.inner.options.command_event_handler.as_ref();
+        if !(tracing_emitter.is_some() || apm_event_handler.is_some()) {
+            return;
+        }
+
+        let event = generate_event();
+        if let (Some(event_handler), Some(ref tracing_emitter)) =
+            (apm_event_handler, &tracing_emitter)
+        {
+            handle_command_event(event_handler.as_ref(), event.clone());
+            handle_command_event(tracing_emitter, event);
+        } else if let Some(event_handler) = apm_event_handler {
+            handle_command_event(event_handler.as_ref(), event);
+        } else if let Some(ref tracing_emitter) = tracing_emitter {
+            handle_command_event(tracing_emitter, event);
         }
     }
 

src/client/options/mod.rs

@@ -530,6 +530,20 @@ pub struct ClientOptions {
     #[builder(default)]
     pub tls: Option<Tls>,
 
+    /// The maximum number of bytes that the driver should include in a tracing event
+    /// or log message's extended JSON string representation of a BSON document, e.g. a
+    /// command or reply from the server.
+    /// If truncation of a document at the exact specified length would occur in the middle
+    /// of a Unicode codepoint, the document will be truncated at the closest larger length
+    /// which falls on a boundary between codepoints.
+    /// Note that in cases where truncation occurs the output will not be valid JSON.
+    ///
+    /// The default value is 1000.
+    #[cfg(any(feature = "tracing-unstable", docsrs))]
+    #[cfg_attr(docsrs, doc(cfg(feature = "tracing-unstable")))]
+    #[builder(default)]
+    pub tracing_max_document_length_bytes: Option<usize>,
+
     /// Specifies the default write concern for operations performed on the Client. See the
     /// WriteConcern type documentation for more details.
     #[builder(default)]
@@ -1259,6 +1273,8 @@ impl ClientOptions {
             sdam_event_handler: None,
             #[cfg(test)]
             test_options: None,
+            #[cfg(feature = "tracing-unstable")]
+            tracing_max_document_length_bytes: None,
         }
     }
 

src/client/session/test/causal_consistency.rs

@@ -6,8 +6,9 @@ use crate::{
     client::options::SessionOptions,
     coll::options::CollectionOptions,
     error::Result,
+    event::command::CommandEvent,
     options::ReadConcern,
-    test::{log_uncaptured, CommandEvent, EventClient, LOCK},
+    test::{log_uncaptured, EventClient, LOCK},
     ClientSession,
     Collection,
 };