vis: add more streamed events

Author: dyc3

crates/ott-balancer/src/client.rs

@@ -128,7 +128,7 @@ pub async fn client_entry<'r>(
     let client_id = Uuid::new_v4().into();
     let client = UnauthorizedClient {
         id: client_id,
-        room: room_name,
+        room: room_name.clone(),
     };
     tracing::Span::current().record("client_id", client_id.to_string());
     info!("client connected");
@@ -199,7 +199,7 @@ pub async fn client_entry<'r>(
         tokio::select! {
             Ok(msg) = client_link.outbound_recv() => {
                 if let SocketMessage::Message(msg) = msg {
-                    debug!(event = "ws", node_id = %client_id, direction = "tx");
+                    debug!(event = "ws", node_id = %client_id, room = %room_name, direction = "tx");
                     if let Err(err) = stream.send(msg).await {
                         error!("Error sending ws message to client: {:?}", err);
                         break;
@@ -220,7 +220,7 @@ pub async fn client_entry<'r>(
                         continue;
                     }
 
-                    debug!(event = "ws", node_id = %client_id, direction = "rx");
+                    debug!(event = "ws", node_id = %client_id, room = %room_name, direction = "rx");
                     if let Err(err) = client_link.inbound_send(msg).await {
                         error!("Error sending client message to balancer: {:?}", err);
                         break;

crates/ott-balancer/src/monolith.rs

@@ -6,7 +6,7 @@ use ott_balancer_protocol::monolith::*;
 use ott_balancer_protocol::*;
 use ott_common::discovery::ConnectionConfig;
 use tokio_tungstenite::tungstenite::Message;
-use tracing::error;
+use tracing::{debug, error};
 
 use crate::messages::*;
 
@@ -194,6 +194,7 @@ impl Room {
 
     /// Broadcast a message to all clients in this room.
     pub fn broadcast(&self, msg: impl Into<SocketMessage>) -> anyhow::Result<()> {
+        debug!(event = "broadcast", node_id = %self.name, direction = "tx");
         self.broadcast_tx.send(msg.into())?;
         Ok(())
     }

crates/ott-balancer/src/service.rs

@@ -246,6 +246,7 @@ impl Service<Request<IncomingBody>> for BalancerService {
                             };
                         if let Some(monolith) = monolith {
                             info!("proxying request to monolith {}", monolith.id());
+                            debug!(event = "proxy", direction = "tx", room = %room_name, node_id = %monolith.id());
                             match proxy_request(req, monolith).await {
                                 Ok(res) => Ok(res),
                                 Err(err) => {
@@ -266,6 +267,7 @@ impl Service<Request<IncomingBody>> for BalancerService {
                             message = "proxying request to monolith",
                             monolith = %monolith.id(),
                         );
+                        debug!(event = "proxy", direction = "tx", node_id = %monolith.id());
                         match proxy_request(req, monolith).await {
                             Ok(res) => Ok(res),
                             Err(err) => {

crates/ott-collector/src/collector.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use once_cell::sync::Lazy;
-use ott_balancer_protocol::collector::BalancerState;
+use ott_balancer_protocol::{collector::BalancerState, RoomName};
 use ott_common::discovery::{ConnectionConfig, ServiceDiscoveryMsg};
 use rocket::futures::StreamExt;
 use serde::Deserialize;
@@ -205,6 +205,8 @@ fn should_send(event: &str) -> bool {
 #[non_exhaustive]
 enum Event {
     Ws(EventWebsocketMessage),
+    Proxy(EventProxyRequest),
+    Broadcast(EventBroadcast),
 }
 
 /// Indicates that a message was sent or received on a websocket connection to a balancer.
@@ -213,6 +215,22 @@ enum Event {
 struct EventWebsocketMessage {
     node_id: Uuid,
     direction: MsgDirection,
+    room: Option<RoomName>,
+}
+
+#[derive(Debug, Deserialize)]
+#[allow(dead_code)]
+struct EventProxyRequest {
+    node_id: Uuid,
+    direction: MsgDirection,
+    room: Option<RoomName>,
+}
+
+#[derive(Debug, Deserialize)]
+#[allow(dead_code)]
+struct EventBroadcast {
+    node_id: String,
+    direction: MsgDirection,
 }
 
 #[derive(Debug, Deserialize)]
@@ -228,7 +246,7 @@ mod tests {
     use uuid::uuid;
 
     #[test]
-    fn test_deserialize_event() {
+    fn test_deserialize_event_ws() {
         let event =
             r#"{"event":"ws","node_id":"f47ac10b-58cc-4372-a567-0e02b2c3d479", "direction": "tx"}"#;
         let event: Event = serde_json::from_str(event).unwrap();
@@ -242,4 +260,66 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn test_deserialize_event_ws2() {
+        let event = r#"{"event":"ws","node_id":"f47ac10b-58cc-4372-a567-0e02b2c3d479", "room": "foo", "direction": "tx"}"#;
+        let event: Event = serde_json::from_str(event).unwrap();
+        #[allow(unreachable_patterns)]
+        match event {
+            Event::Ws(EventWebsocketMessage { node_id, room, .. }) => {
+                assert_eq!(node_id, uuid!("f47ac10b-58cc-4372-a567-0e02b2c3d479"));
+                assert_eq!(room, Some("foo".into()));
+            }
+            _ => {
+                panic!("unexpected event type: {:?}", event);
+            }
+        }
+    }
+
+    #[test]
+    fn test_deserialize_event_proxy() {
+        let event = r#"{"event":"proxy","node_id":"f47ac10b-58cc-4372-a567-0e02b2c3d479", "direction": "tx"}"#;
+        let event: Event = serde_json::from_str(event).unwrap();
+        #[allow(unreachable_patterns)]
+        match event {
+            Event::Proxy(EventProxyRequest { node_id, .. }) => {
+                assert_eq!(node_id, uuid!("f47ac10b-58cc-4372-a567-0e02b2c3d479"));
+            }
+            _ => {
+                panic!("unexpected event type: {:?}", event);
+            }
+        }
+    }
+
+    #[test]
+    fn test_deserialize_event_proxy2() {
+        let event = r#"{"event":"proxy","node_id":"f47ac10b-58cc-4372-a567-0e02b2c3d479", "room": "foo", "direction": "tx"}"#;
+        let event: Event = serde_json::from_str(event).unwrap();
+        #[allow(unreachable_patterns)]
+        match event {
+            Event::Proxy(EventProxyRequest { node_id, room, .. }) => {
+                assert_eq!(node_id, uuid!("f47ac10b-58cc-4372-a567-0e02b2c3d479"));
+                assert_eq!(room, Some("foo".into()));
+            }
+            _ => {
+                panic!("unexpected event type: {:?}", event);
+            }
+        }
+    }
+
+    #[test]
+    fn test_deserialize_event_broadcast() {
+        let event = r#"{"event":"broadcast","node_id":"foo", "direction": "tx"}"#;
+        let event: Event = serde_json::from_str(event).unwrap();
+        #[allow(unreachable_patterns)]
+        match event {
+            Event::Broadcast(EventBroadcast { node_id, .. }) => {
+                assert_eq!(node_id, "foo");
+            }
+            _ => {
+                panic!("unexpected event type: {:?}", event);
+            }
+        }
+    }
 }

packages/ott-vis-datasource/src/datasource.ts

@@ -37,6 +37,7 @@ export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
 					frame.addField({ name: "event", type: FieldType.string });
 					frame.addField({ name: "node_id", type: FieldType.string });
 					frame.addField({ name: "direction", type: FieldType.string });
+					frame.addField({ name: "room", type: FieldType.string });
 
 					const base = this.baseUrl.replace(/^http/, "ws");
 

packages/ott-vis-panel/src/components/views/TopologyView.tsx

@@ -17,7 +17,7 @@ import "./topology-view.css";
 import { useColorProvider } from "colors";
 import { useD3Zoom } from "chartutils";
 import { dedupeItems } from "aggregate";
-import { useEventBus } from "eventbus";
+import { useEventBus, type BusEvent } from "eventbus";
 
 /**
  * The goal of this component is to show a more accurate topology view from the perspective of actual network connections.
@@ -410,6 +410,7 @@ export const TopologyView: React.FC<TopologyViewProps> = ({
 	useEffect(() => {
 		const rxColor = "#0f0";
 		const txColor = "#00f";
+		const proxyColor = "#f58d05";
 
 		function animateNode(
 			node: d3.Selection<any, d3.HierarchyNode<TreeNode>, any, any>,
@@ -461,8 +462,18 @@ export const TopologyView: React.FC<TopologyViewProps> = ({
 				.attrTween("stroke-width", () => t => d3.interpolateNumber(4, 1.5)(t).toString());
 		}
 
+		function getColor(event: BusEvent): string {
+			if (event.event === "ws" || event.event === "broadcast") {
+				return event.direction === "rx" ? rxColor : txColor;
+			} else if (event.event === "proxy") {
+				return proxyColor;
+			} else {
+				return "#f00";
+			}
+		}
+
 		const sub = eventBus.subscribe(event => {
-			const color = event.direction === "rx" ? rxColor : txColor;
+			const color = getColor(event);
 			const node = d3.select<d3.BaseType, d3.HierarchyNode<TreeNode>>(
 				`[data-nodeid="${event.node_id}"]`
 			);
@@ -478,6 +489,18 @@ export const TopologyView: React.FC<TopologyViewProps> = ({
 				`[data-nodeid-target="${event.node_id}"]`
 			);
 			animateLinks(links, color);
+
+			if (event.room) {
+				const room = d3.select<d3.BaseType, d3.HierarchyNode<TreeNode>>(
+					`[data-nodeid="${event.room}"]`
+				);
+				animateNode(room, color);
+
+				const roomLink = d3.select<d3.BaseType, d3.HierarchyLink<TreeNode>>(
+					`[data-nodeid-target="${event.room}"]`
+				);
+				animateLinks(roomLink, color);
+			}
 		});
 
 		return () => {

packages/ott-vis-panel/src/eventbus.ts

@@ -11,4 +11,5 @@ export type BusEvent = {
 	event: string;
 	node_id: string;
 	direction: "tx" | "rx";
+	room?: string;
 };