Tweak logging around connections opening and messages coming in

backend/telemetry_core/src/aggregator/inner_loop.rs

@@ -224,8 +224,6 @@ impl InnerLoop {
 
     /// Handle messages coming from shards.
     fn handle_from_shard(&mut self, shard_conn_id: ConnId, msg: FromShardWebsocket) {
-        log::debug!("Message from shard ({:?}): {:?}", shard_conn_id, msg);
-
         match msg {
             FromShardWebsocket::Initialize { channel } => {
                 self.shard_channels.insert(shard_conn_id, channel);
@@ -357,7 +355,6 @@ impl InnerLoop {
 
     /// Handle messages coming from feeds.
     fn handle_from_feed(&mut self, feed_conn_id: ConnId, msg: FromFeedWebsocket) {
-        log::debug!("Message from feed ({:?}): {:?}", feed_conn_id, msg);
         match msg {
             FromFeedWebsocket::Initialize { channel } => {
                 self.feed_channels.insert(feed_conn_id, channel.clone());

backend/telemetry_core/src/main.rs

@@ -105,6 +105,7 @@ async fn start_server(opts: Opts) -> anyhow::Result<()> {
                 (&Method::GET, "/health") => Ok(Response::new("OK".into())),
                 // Subscribe to feed messages:
                 (&Method::GET, "/feed") => {
+                    log::info!("Opening /feed connection from {:?}", addr);
                     Ok(http_utils::upgrade_to_websocket(
                         req,
                         move |ws_send, ws_recv| async move {
@@ -130,6 +131,7 @@ async fn start_server(opts: Opts) -> anyhow::Result<()> {
                     Ok(http_utils::upgrade_to_websocket(
                         req,
                         move |ws_send, ws_recv| async move {
+                            log::info!("Opening /shard_submit connection from {:?}", addr);
                             let tx_to_aggregator = aggregator.subscribe_shard();
                             let (mut tx_to_aggregator, mut ws_send) =
                                 handle_shard_websocket_connection(
@@ -399,10 +401,6 @@ let mut i: u64 = 0;
 if feed_id == 1 {
     i += 1;
     println!("FEED #{}, msgs: {}", i, msgs.len());
-    if i > 1000 {
-        log::error!("TESTING: close feed");
-        break
-    }
 }
             // End the loop when there are more than 10k messages queued up.
             // This number is just picked as a fairly high limit that should account

backend/telemetry_shard/src/main.rs

@@ -138,6 +138,7 @@ async fn start_server(opts: Opts) -> anyhow::Result<()> {
                     Ok(http_utils::upgrade_to_websocket(
                         req,
                         move |ws_send, ws_recv| async move {
+                            log::info!("Opening /submit connection from {:?}", addr);
                             let tx_to_aggregator = aggregator.subscribe_node();
                             let (mut tx_to_aggregator, mut ws_send) =
                                 handle_node_websocket_connection(