update relation graph

Author: fuyufjh

dashboard/components/RelationGraph.tsx

@@ -40,6 +40,7 @@ import {
   epochToUnixMillis,
   latencyToColor,
 } from "./utils/backPressure"
+import { ChannelStatsDerived } from "../pages/fragment_graph"
 
 // Size of each relation box in pixels
 export const boxWidth = 150
@@ -74,13 +75,13 @@ export default function RelationGraph({
   nodes,
   selectedId,
   setSelectedId,
-  backPressures,
+  channelStats,
   relationStats,
 }: {
   nodes: RelationPoint[] // rename to RelationNode
   selectedId: string | undefined
   setSelectedId: (id: string) => void
-  backPressures?: Map<string, number> // relationId-relationId->back_pressure_rate})
+  channelStats?: Map<string, ChannelStatsDerived>
   relationStats: { [relationId: number]: RelationStats } | undefined
 }) {
   const [modalData, setModalId] = useCatalogModal(nodes.map((n) => n.relation))
@@ -173,21 +174,21 @@ export default function RelationGraph({
 
     const applyEdge = (sel: EdgeSelection) => {
       const color = (d: Edge) => {
-        if (backPressures) {
-          let value = backPressures.get(`${d.source}_${d.target}`)
+        if (channelStats) {
+          let value = channelStats.get(`${d.source}_${d.target}`)
           if (value) {
-            return backPressureColor(value)
+            return backPressureColor(value.backPressure)
           }
         }
 
         return theme.colors.gray["300"]
       }
 
       const width = (d: Edge) => {
-        if (backPressures) {
-          let value = backPressures.get(`${d.source}_${d.target}`)
+        if (channelStats) {
+          let value = channelStats.get(`${d.source}_${d.target}`)
           if (value) {
-            return backPressureWidth(value, 15)
+            return backPressureWidth(value.backPressure, 15)
           }
         }
         return 2
@@ -208,11 +209,15 @@ export default function RelationGraph({
           d3.selectAll(".tooltip").remove()
 
           // Create new tooltip
-          const bpValue = backPressures?.get(`${d.source}_${d.target}`)
+          const stats = channelStats?.get(`${d.source}_${d.target}`)
           const tooltipText = `<b>Relation ${d.source} → ${
             d.target
           }</b><br>Backpressure: ${
-            bpValue != null ? `${(bpValue * 100).toFixed(2)}%` : "N/A"
+            stats != null ? `${(stats.backPressure * 100).toFixed(2)}%` : "N/A"
+          }<br>Input Throughput: ${
+            stats != null ? `${stats.inputThroughput.toFixed(2)} rows/s` : "N/A"
+          }<br>Output Throughput: ${
+            stats != null ? `${stats.outputThroughput.toFixed(2)} rows/s` : "N/A"
           }`
           d3.select("body")
             .append("div")
@@ -421,7 +426,7 @@ export default function RelationGraph({
     selectedId,
     setModalId,
     setSelectedId,
-    backPressures,
+    channelStats,
     relationStats,
   ])
 

dashboard/pages/relation_graph.tsx

@@ -34,7 +34,8 @@ import {
 } from "../lib/api/streaming"
 import { RelationPoint } from "../lib/layout"
 import { RelationStats } from "../proto/gen/monitor_service"
-import { ChannelStatsSnapshot } from "./fragment_graph"
+import { ChannelStatsDerived, ChannelStatsSnapshot } from "./fragment_graph"
+import { GetBackPressureResponse } from "../proto/gen/monitor_service"
 
 const SIDEBAR_WIDTH = "200px"
 const INTERVAL_MS = 5000
@@ -96,8 +97,8 @@ export default function StreamingGraph() {
   const relationDependency = relationDependencyCallback()
 
   // Periodically fetch fragment-level back-pressure from Meta node
-  const [backPressureRate, setBackPressureRate] =
-    useState<Map<string, number>>()
+  const [channelStats, setChannelStats] = 
+    useState<Map<string, ChannelStatsDerived>>()
   const [relationStats, setRelationStats] = useState<{
     [key: number]: RelationStats
   }>()
@@ -108,20 +109,22 @@ export default function StreamingGraph() {
     let initialSnapshot: ChannelStatsSnapshot | undefined
 
     if (resetEmbeddedBackPressures) {
-      setBackPressureRate(undefined)
+      setChannelStats(undefined)
       toggleResetEmbeddedBackPressures()
     }
 
     function refresh() {
       api.get("/metrics/fragment/embedded_back_pressures").then(
-        (response) => {
-          let snapshot = ChannelStatsSnapshot.fromResponse(
-            response.channelStats
+        (res) => {
+          let response = GetBackPressureResponse.fromJSON(res)
+          let snapshot = new ChannelStatsSnapshot(
+            new Map(Object.entries(response.channelStats)),
+            Date.now()
           )
           if (!initialSnapshot) {
             initialSnapshot = snapshot
           } else {
-            setBackPressureRate(snapshot.getRate(initialSnapshot!))
+            setChannelStats(snapshot.getRate(initialSnapshot))
           }
           setRelationStats(response.relationStats)
         },
@@ -139,26 +142,26 @@ export default function StreamingGraph() {
   }, [toast, resetEmbeddedBackPressures])
 
   // Convert fragment-level backpressure rate map to relation-level backpressure rate
-  const backPressures: Map<string, number> | undefined = useMemo(() => {
+  const relationChannelStats: Map<string, ChannelStatsDerived> | undefined = useMemo(() => {
     if (!fragmentVertexToRelationMap) {
-      return new Map<string, number>()
+      return new Map<string, ChannelStatsDerived>()
     }
     let inMap = fragmentVertexToRelationMap.inMap
     let outMap = fragmentVertexToRelationMap.outMap
-    if (backPressureRate) {
-      let map = new Map<string, number>()
-      for (const [key, value] of backPressureRate) {
+    if (channelStats) {
+      let map = new Map<string, ChannelStatsDerived>()
+      for (const [key, stats] of channelStats) {
         const [outputFragment, inputFragment] = key.split("_").map(Number)
         if (outMap[outputFragment] && inMap[inputFragment]) {
           const outputRelation = outMap[outputFragment]
           const inputRelation = inMap[inputFragment]
           let key = `${outputRelation}_${inputRelation}`
-          map.set(key, value)
+          map.set(key, stats)
         }
       }
       return map
     }
-  }, [backPressureRate, fragmentVertexToRelationMap])
+  }, [channelStats, fragmentVertexToRelationMap])
 
   const retVal = (
     <Flex p={3} height="calc(100vh - 20px)" flexDirection="column">
@@ -214,7 +217,7 @@ export default function StreamingGraph() {
               nodes={relationDependency}
               selectedId={selectedId?.toString()}
               setSelectedId={(id) => setSelectedId(parseInt(id))}
-              backPressures={backPressures}
+              channelStats={relationChannelStats}
               relationStats={relationStats}
             />
           )}