Add `tracing` events to signing and event streams (#2057)

* Emit a trace with the string to sign when signing
* Add traces to event stream message send/receive
* Add a message to dispatch trace

CHANGELOG.next.toml

@@ -689,3 +689,15 @@ message = "Fixed and improved the request `tracing` span hierarchy to improve lo
 references = ["smithy-rs#2044", "smithy-rs#371"]
 meta = { "breaking" = false, "tada" = true, "bug" = false, "target" = "client"}
 author = "jdisanti"
+
+[[aws-sdk-rust]]
+message = "Add more `tracing` events to signing and event streams"
+references = ["smithy-rs#2057", "smithy-rs#371"]
+meta = { "breaking" = false, "tada" = false, "bug" = false }
+author = "jdisanti"
+
+[[smithy-rs]]
+message = "Add more `tracing` events to signing and event streams"
+references = ["smithy-rs#2057", "smithy-rs#371"]
+meta = { "breaking" = false, "tada" = false, "bug" = false, "target" = "client"}
+author = "jdisanti"

aws/rust-runtime/aws-sigv4/src/event_stream.rs

@@ -126,6 +126,7 @@ fn sign_payload<'a>(
         params,
     );
     let signature = calculate_signature(signing_key, &string_to_sign);
+    tracing::trace!(canonical_request = ?message_payload, string_to_sign = ?string_to_sign, "calculated signing parameters");
 
     // Generate the signed wrapper event frame
     SigningOutput::new(

aws/rust-runtime/aws-sigv4/src/http_request/sign.rs

@@ -180,22 +180,23 @@ fn calculate_signing_params<'a>(
     params: &'a SigningParams<'a>,
 ) -> Result<(CalculatedParams, String), SigningError> {
     let creq = CanonicalRequest::from(request, params)?;
-    tracing::trace!(canonical_request = %creq);
 
     let encoded_creq = &sha256_hex_string(creq.to_string().as_bytes());
-    let sts = StringToSign::new(
+    let string_to_sign = StringToSign::new(
         params.time,
         params.region,
         params.service_name,
         encoded_creq,
-    );
+    )
+    .to_string();
     let signing_key = generate_signing_key(
         params.secret_key,
         params.time,
         params.region,
         params.service_name,
     );
-    let signature = calculate_signature(signing_key, sts.to_string().as_bytes());
+    let signature = calculate_signature(signing_key, string_to_sign.as_bytes());
+    tracing::trace!(canonical_request = %creq, string_to_sign = %string_to_sign, "calculated signing parameters");
 
     let values = creq.values.into_query_params().expect("signing with query");
     let mut signing_params = vec![

rust-runtime/aws-smithy-http-tower/src/dispatch.rs

@@ -44,7 +44,7 @@ where
         let (req, property_bag) = req.into_parts();
         let mut inner = self.inner.clone();
         let future = async move {
-            trace!(request = ?req);
+            trace!(request = ?req, "dispatching request");
             inner
                 .call(req)
                 .await

rust-runtime/aws-smithy-http/src/event_stream/receiver.rs

@@ -16,6 +16,7 @@ use std::error::Error as StdError;
 use std::fmt;
 use std::marker::PhantomData;
 use std::mem;
+use tracing::trace;
 
 /// Wrapper around SegmentedBuf that tracks the state of the stream.
 #[derive(Debug)]
@@ -198,6 +199,7 @@ impl<T, E> Receiver<T, E> {
                         )
                     })?
                 {
+                    trace!(message = ?message, "received complete event stream message");
                     return Ok(Some(message));
                 }
             }
@@ -205,6 +207,7 @@ impl<T, E> Receiver<T, E> {
             self.buffer_next_chunk().await?;
         }
         if self.buffer.has_data() {
+            trace!(remaining_data = ?self.buffer, "data left over in the event stream response stream");
             return Err(SdkError::response_error(
                 ReceiverError {
                     kind: ReceiverErrorKind::UnexpectedEndOfStream,

rust-runtime/aws-smithy-http/src/event_stream/sender.rs

@@ -13,6 +13,7 @@ use std::fmt::Debug;
 use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use tracing::trace;
 
 /// Input type for Event Streams.
 pub struct EventStreamSender<T, E> {
@@ -150,14 +151,18 @@ impl<T, E: StdError + Send + Sync + 'static> Stream for MessageStreamAdapter<T,
                             .marshall(message)
                             .map_err(SdkError::construction_failure)?,
                     };
+
+                    trace!(unsigned_message = ?message, "signing event stream message");
                     let message = self
                         .signer
                         .sign(message)
                         .map_err(SdkError::construction_failure)?;
+
                     let mut buffer = Vec::new();
                     message
                         .write_to(&mut buffer)
                         .map_err(SdkError::construction_failure)?;
+                    trace!(signed_message = ?buffer, "sending signed event stream message");
                     Poll::Ready(Some(Ok(Bytes::from(buffer))))
                 } else if !self.end_signal_sent {
                     self.end_signal_sent = true;
@@ -167,6 +172,7 @@ impl<T, E: StdError + Send + Sync + 'static> Stream for MessageStreamAdapter<T,
                             sign.map_err(SdkError::construction_failure)?
                                 .write_to(&mut buffer)
                                 .map_err(SdkError::construction_failure)?;
+                            trace!(signed_message = ?buffer, "sending signed empty message to terminate the event stream");
                             Poll::Ready(Some(Ok(Bytes::from(buffer))))
                         }
                         None => Poll::Ready(None),