Don't panic on debug when packets aren't utf-8

src/lib.rs

@@ -27,6 +27,7 @@ pub mod extensions;
 pub mod metrics;
 pub mod proxy;
 pub mod test_utils;
+pub(crate) mod utils;
 pub(crate) mod xds;
 
 #[cfg(doctest)]

src/proxy/server/mod.rs

@@ -16,24 +16,25 @@
 
 use std::collections::HashMap;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
-use std::str::from_utf8;
 use std::sync::Arc;
 
-use slog::{debug, error, info, warn, Logger};
+use slog::{debug, error, info, trace, warn, Logger};
 use tokio::net::udp::{RecvHalf, SendHalf};
 use tokio::net::UdpSocket;
 use tokio::sync::{mpsc, watch};
 use tokio::sync::{Mutex, RwLock};
 use tokio::time::{delay_for, Duration, Instant};
 
+use metrics::Metrics as ProxyMetrics;
+
+use crate::cluster::cluster_manager::{ClusterManager, SharedClusterManager};
 use crate::config::{Config, EndPoint, Source};
 use crate::extensions::{DownstreamContext, Filter, FilterChain};
+use crate::proxy::server::error::{Error, RecvFromError};
 use crate::proxy::sessions::{Packet, Session, SESSION_TIMEOUT_SECONDS};
+use crate::utils::debug;
 
 use super::metrics::{start_metrics_server, Metrics};
-use crate::cluster::cluster_manager::{ClusterManager, SharedClusterManager};
-use crate::proxy::server::error::{Error, RecvFromError};
-use metrics::Metrics as ProxyMetrics;
 
 pub mod error;
 pub(super) mod metrics;
@@ -203,11 +204,11 @@ impl Server {
         tokio::spawn(async move {
             let packet = &buf[..size];
 
-            debug!(
+            trace!(
                 args.log,
-                "Packet Received from: {}, {}",
-                recv_addr,
-                from_utf8(packet).unwrap()
+                "Packet Received";
+                "from" => recv_addr,
+                "contents" => debug::bytes_to_string(packet.to_vec()),
             );
 
             let endpoints = match args.cluster_manager.read().get_all_endpoints() {
@@ -282,7 +283,7 @@ impl Server {
                     log,
                     "Sending packet back to origin";
                     "origin" => packet.dest(),
-                    "contents" => String::from_utf8(packet.contents().clone()).unwrap(),
+                    "contents" => debug::bytes_to_string(packet.contents().clone()),
                 );
 
                 if let Err(err) = send_socket
@@ -384,14 +385,14 @@ mod tests {
 
     use crate::config;
     use crate::config::{Builder as ConfigBuilder, EndPoint, ProxyMode};
+    use crate::extensions::FilterRegistry;
     use crate::proxy::sessions::{Packet, SESSION_TIMEOUT_SECONDS};
+    use crate::proxy::Builder;
     use crate::test_utils::{
         config_with_dummy_endpoint, SplitSocket, TestFilter, TestFilterFactory, TestHelper,
     };
 
     use super::*;
-    use crate::extensions::FilterRegistry;
-    use crate::proxy::Builder;
 
     #[tokio::test]
     async fn run_server() {

src/proxy/sessions/session.rs

@@ -16,12 +16,11 @@
 
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::result;
-use std::str::from_utf8;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering::Relaxed;
 use std::sync::Arc;
 
-use slog::{debug, error, o, Logger};
+use slog::{debug, error, o, trace, Logger};
 use tokio::net::udp::{RecvHalf, SendHalf};
 use tokio::net::UdpSocket;
 use tokio::select;
@@ -30,9 +29,9 @@ use tokio::time::{Duration, Instant};
 
 use crate::config::EndPoint;
 use crate::extensions::{Filter, FilterChain, UpstreamContext};
-
 use crate::proxy::sessions::error::Error;
 use crate::proxy::sessions::metrics::Metrics;
+use crate::utils::debug;
 
 type Result<T> = std::result::Result<T, Error>;
 
@@ -213,7 +212,10 @@ impl Session {
             from,
             to,
         } = packet_ctx;
-        debug!(log, "Received packet"; "from" => from, "endpoint_name" => &endpoint.name, "endpoint_addr" => &endpoint.address, "contents" => from_utf8(packet).unwrap());
+
+        trace!(log, "Received packet"; "from" => from, "endpoint_name" => &endpoint.name,
+            "endpoint_addr" => &endpoint.address, 
+            "contents" => debug::bytes_to_string(packet.to_vec()));
         Session::inc_expiration(expiration).await;
 
         if let Some(response) =
@@ -242,7 +244,9 @@ impl Session {
 
     /// Sends a packet to the Session's dest.
     pub async fn send_to(&mut self, buf: &[u8]) -> Result<Option<usize>> {
-        debug!(self.log, "Sending packet"; "dest_name" => &self.dest.name, "dest_address" => &self.dest.address, "contents" => from_utf8(buf).unwrap());
+        trace!(self.log, "Sending packet"; "dest_name" => &self.dest.name,
+        "dest_address" => &self.dest.address, 
+        "contents" => debug::bytes_to_string(buf.to_vec()));
 
         self.send
             .send_to(buf, &self.dest.address)
@@ -282,6 +286,8 @@ impl Drop for Session {
 
 #[cfg(test)]
 mod tests {
+    use std::str::from_utf8;
+
     use prometheus::Registry;
     use slog::info;
     use tokio::time;

src/utils/debug.rs

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2020 Google LLC All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+//!
+//! The `debug` module is for functionality related to realtime debugging of this project.
+//!
+
+/// Attempt to convert a packet into a string, if it is one, otherwise return some human
+/// readable details about the packet.
+pub(crate) fn bytes_to_string(bytes: Vec<u8>) -> String {
+    String::from_utf8(bytes.to_vec())
+        .unwrap_or_else(|_| format!("<raw bytes :: len: {}>", bytes.len()))
+}

src/utils/mod.rs

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2020 Google LLC All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+pub(crate) mod debug;