Improve janus related errors logging

src/backend/janus/client.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::fmt;
 use std::sync::Arc;
 use std::thread;
 use std::time::Duration as StdDuration;
@@ -27,6 +28,18 @@ struct RequestInfo {
     payload: JsonValue,
 }
 
+impl fmt::Display for RequestInfo {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        let method = self
+            .payload
+            .get("method")
+            .and_then(|m| m.as_str())
+            .unwrap_or("");
+
+        write!(f, "to = '{}', method = '{}'", self.to, method)
+    }
+}
+
 enum TransactionWatchdogMessage {
     Halt,
     Insert(String, RequestInfo),
@@ -67,12 +80,11 @@ impl Client {
 
                 state = state
                     .into_iter()
-                    .filter(|(corr_data, info)| {
+                    .filter(|(_corr_data, info)| {
                         if info.start_timestamp + info.timeout < Utc::now() {
-                            let err =
-                                anyhow!("Janus request timed out ({}): {:?}", corr_data, info);
-
+                            let err = anyhow!("Janus request timed out; {}", info);
                             error!(crate::LOG, "{}", err);
+
                             if let Some(ref writer) = timeouts_writer {
                                 writer.record_janus_timeout(info.to.clone());
                             }

src/backend/janus/mod.rs

@@ -217,6 +217,8 @@ async fn handle_response_impl<C: Context>(
                     inresp
                         .plugin()
                         .data()
+                        .ok_or_else(|| anyhow!("Missing 'data' in the response"))
+                        .error(AppErrorKind::MessageParsingFailed)?
                         .get("status")
                         .ok_or_else(|| anyhow!("Missing 'status' in the response"))
                         .error(AppErrorKind::MessageParsingFailed)
@@ -260,6 +262,8 @@ async fn handle_response_impl<C: Context>(
                     inresp
                         .plugin()
                         .data()
+                        .ok_or_else(|| anyhow!("Missing 'data' in the response"))
+                        .error(AppErrorKind::MessageParsingFailed)?
                         .get("status")
                         .ok_or_else(|| anyhow!("Missing 'status' in the response"))
                         .error(AppErrorKind::MessageParsingFailed)
@@ -305,7 +309,11 @@ async fn handle_response_impl<C: Context>(
                     ));
 
                     // TODO: improve error handling
-                    let plugin_data = inresp.plugin().data();
+                    let plugin_data = inresp
+                        .plugin()
+                        .data()
+                        .ok_or_else(|| anyhow!("Missing 'data' in the response"))
+                        .error(AppErrorKind::MessageParsingFailed)?;
 
                     plugin_data
                         .get("status")
@@ -453,15 +461,15 @@ async fn handle_response_impl<C: Context>(
         }
         IncomingResponse::Error(ErrorResponse::Session(ref inresp)) => {
             let err = anyhow!(
-                "received an unexpected Error message (session): {:?}",
-                inresp
+                "received an unexpected Error message (session): {}",
+                inresp.error()
             );
             Err(err).error(AppErrorKind::MessageParsingFailed)
         }
         IncomingResponse::Error(ErrorResponse::Handle(ref inresp)) => {
             let err = anyhow!(
-                "received an unexpected Error message (handle): {:?}",
-                inresp
+                "received an unexpected Error message (handle): {}",
+                inresp.error()
             );
             Err(err).error(AppErrorKind::MessageParsingFailed)
         }

src/backend/janus/responses.rs

@@ -1,3 +1,5 @@
+use std::fmt;
+
 use serde_derive::Deserialize;
 use serde_json::Value as JsonValue;
 
@@ -26,18 +28,36 @@ pub(crate) struct HandleErrorResponse {
     error: ErrorResponseData,
 }
 
+impl HandleErrorResponse {
+    pub(crate) fn error(&self) -> &ErrorResponseData {
+        &self.error
+    }
+}
+
 #[derive(Debug, Deserialize)]
 pub(crate) struct SessionErrorResponse {
     transaction: String,
     error: ErrorResponseData,
 }
 
+impl SessionErrorResponse {
+    pub(crate) fn error(&self) -> &ErrorResponseData {
+        &self.error
+    }
+}
+
 #[derive(Debug, Deserialize)]
 pub(crate) struct ErrorResponseData {
     code: i32,
     reason: String,
 }
 
+impl fmt::Display for ErrorResponseData {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{} ({})", self.reason, self.code)
+    }
+}
+
 // A request to a plugin handle was received.
 #[derive(Debug, Deserialize)]
 pub(crate) struct AckResponse {
@@ -77,13 +97,13 @@ impl EventResponse {
 
 #[derive(Debug, Deserialize)]
 pub(crate) struct EventResponsePluginData {
-    data: JsonValue,
+    data: Option<JsonValue>,
     plugin: String,
 }
 
 impl EventResponsePluginData {
-    pub(crate) fn data(&self) -> &JsonValue {
-        &self.data
+    pub(crate) fn data(&self) -> Option<&JsonValue> {
+        self.data.as_ref()
     }
 }