From 50c856c9c22e3e1794e5ab16c10f52922ad8abca Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Fri, 1 Mar 2024 17:07:57 +0800
Subject: [PATCH 01/13] remove the p50/90/99 backpressure

---
 dashboard/lib/api/metric.ts        | 10 ++++-
 dashboard/pages/fragment_graph.tsx | 59 +++---------------------------
 2 files changed, 14 insertions(+), 55 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index 1b36be23e0d02..a155909eb0d65 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -125,7 +125,7 @@ export function calculateBPRate(
         key,
         // The *100 in end of the formular is to convert the BP rate to the value used in web UI drawing
         ((value - (mapOld.get(key) || 0)) / ((INTERVAL / 1000) * 1000000000)) *
-          100
+        100
       )
     } else {
       result.set(key, 0)
@@ -180,6 +180,14 @@ export function p99(samples: MetricsSample[]) {
   return calculatePercentile(samples, 0.99)
 }
 
+export function average(samples: MetricsSample[]) {
+  let sum = 0
+  for (const sample of samples) {
+    sum += sample.value
+  }
+  return sum / samples.length
+}
+
 function isSet(value: any): boolean {
   return value !== null && value !== undefined
 }
diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index e5179e69fdce4..ad9806643352a 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -42,13 +42,10 @@ import {
   BackPressureInfo,
   BackPressuresMetrics,
   INTERVAL,
+  average,
   calculateBPRate,
   getActorBackPressures,
   getBackPressureWithoutPrometheus,
-  p50,
-  p90,
-  p95,
-  p99,
 } from "../lib/api/metric"
 import { getFragments, getStreamingJobs } from "../lib/api/streaming"
 import { FragmentBox } from "../lib/layout"
@@ -179,9 +176,6 @@ function buildFragmentDependencyAsEdges(
 
 const SIDEBAR_WIDTH = 200
 
-type BackPressureAlgo = "p50" | "p90" | "p95" | "p99"
-const backPressureAlgos: BackPressureAlgo[] = ["p50", "p90", "p95", "p99"]
-
 type BackPressureDataSource = "Embedded" | "Prometheus"
 const backPressureDataSources: BackPressureDataSource[] = [
   "Embedded",
@@ -193,7 +187,6 @@ export default function Streaming() {
   const { response: fragmentList } = useFetch(getFragments)
 
   const [relationId, setRelationId] = useQueryState("id", parseAsInteger)
-  const [backPressureAlgo, setBackPressureAlgo] = useQueryState("backPressure")
   const [selectedFragmentId, setSelectedFragmentId] = useState<number>()
   // used to get the data source
   const [backPressureDataSourceAlgo, setBackPressureDataSourceAlgo] =
@@ -202,7 +195,7 @@ export default function Streaming() {
   const { response: actorBackPressures } = useFetch(
     getActorBackPressures,
     INTERVAL,
-    backPressureDataSourceAlgo === "Prometheus" && backPressureAlgo !== null
+    backPressureDataSourceAlgo === "Prometheus"
   )
 
   const fragmentDependencyCallback = useCallback(() => {
@@ -230,7 +223,7 @@ export default function Streaming() {
         }
       }
     }
-    return () => {}
+    return () => { }
   }, [relationId, relationList, setRelationId])
 
   // get back pressure rate without prometheus
@@ -330,10 +323,7 @@ export default function Streaming() {
   }
 
   const backPressures = useMemo(() => {
-    if (
-      (actorBackPressures && backPressureAlgo && backPressureAlgo) ||
-      backPressuresMetricsWithoutPromtheus
-    ) {
+    if (actorBackPressures || backPressuresMetricsWithoutPromtheus) {
       let map = new Map()
 
       if (
@@ -351,25 +341,7 @@ export default function Streaming() {
         actorBackPressures
       ) {
         for (const m of actorBackPressures.outputBufferBlockingDuration) {
-          let algoFunc
-          switch (backPressureAlgo) {
-            case "p50":
-              algoFunc = p50
-              break
-            case "p90":
-              algoFunc = p90
-              break
-            case "p95":
-              algoFunc = p95
-              break
-            case "p99":
-              algoFunc = p99
-              break
-            default:
-              return
-          }
-
-          const value = algoFunc(m.sample) * 100
+          const value = average(m.sample) * 100
           map.set(
             `${m.metric.fragment_id}_${m.metric.downstream_fragment_id}`,
             value
@@ -381,7 +353,6 @@ export default function Streaming() {
   }, [
     backPressureDataSourceAlgo,
     actorBackPressures,
-    backPressureAlgo,
     backPressuresMetricsWithoutPromtheus,
   ])
 
@@ -468,26 +439,6 @@ export default function Streaming() {
               ))}
             </Select>
           </FormControl>
-          {backPressureDataSourceAlgo === "Prometheus" && (
-            <FormControl>
-              <FormLabel>Back Pressure Algorithm</FormLabel>
-              <Select
-                value={backPressureAlgo ?? undefined}
-                onChange={(event) => {
-                  setBackPressureAlgo(event.target.value as BackPressureAlgo)
-                }}
-              >
-                <option value="" disabled selected hidden>
-                  Please select
-                </option>
-                {backPressureAlgos.map((algo) => (
-                  <option value={algo} key={algo}>
-                    {algo}
-                  </option>
-                ))}
-              </Select>
-            </FormControl>
-          )}
           <Flex height="full" width="full" flexDirection="column">
             <Text fontWeight="semibold">Fragments</Text>
             {fragmentDependencyDag && (

From 4129e3aa870f27f2c89bfcac40f7854d4b0f3abf Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Fri, 1 Mar 2024 17:09:42 +0800
Subject: [PATCH 02/13] rename

---
 dashboard/lib/api/metric.ts        |  3 +-
 dashboard/pages/fragment_graph.tsx | 49 +++++++++++++++---------------
 2 files changed, 26 insertions(+), 26 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index a155909eb0d65..647419672f7b6 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -17,7 +17,6 @@
 import { Metrics, MetricsSample } from "../../components/metrics"
 import api from "./api"
 
-export const INTERVAL = 5000
 export interface BackPressuresMetrics {
   outputBufferBlockingDuration: Metrics[]
 }
@@ -149,7 +148,7 @@ export const BackPressureInfo = {
 }
 
 // Get back pressure from meta node -> compute node
-export async function getBackPressureWithoutPrometheus() {
+export async function fetchEmbeddedBackPressure() {
   const response = await api.get("/metrics/fragment/embedded_back_pressures")
   let backPressureInfos: BackPressureInfo[] = response.backPressureInfos.map(
     BackPressureInfo.fromJSON
diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index ad9806643352a..c47d00d21674b 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -41,11 +41,10 @@ import useFetch from "../lib/api/fetch"
 import {
   BackPressureInfo,
   BackPressuresMetrics,
-  INTERVAL,
   average,
   calculateBPRate,
   getActorBackPressures,
-  getBackPressureWithoutPrometheus,
+  fetchEmbeddedBackPressure,
 } from "../lib/api/metric"
 import { getFragments, getStreamingJobs } from "../lib/api/streaming"
 import { FragmentBox } from "../lib/layout"
@@ -56,6 +55,9 @@ interface DispatcherNode {
   [actorId: number]: Dispatcher[]
 }
 
+// Refresh interval (ms) for back pressure stats
+const INTERVAL = 5000
+
 /** Associated data of each plan node in the fragment graph, including the dispatchers. */
 export interface PlanNodeDatum {
   name: string
@@ -189,13 +191,13 @@ export default function Streaming() {
   const [relationId, setRelationId] = useQueryState("id", parseAsInteger)
   const [selectedFragmentId, setSelectedFragmentId] = useState<number>()
   // used to get the data source
-  const [backPressureDataSourceAlgo, setBackPressureDataSourceAlgo] =
+  const [backPressureDataSource, setBackPressureDataSource] =
     useState("Embedded")
 
-  const { response: actorBackPressures } = useFetch(
+  const { response: promethusMetrics } = useFetch(
     getActorBackPressures,
     INTERVAL,
-    backPressureDataSourceAlgo === "Prometheus"
+    backPressureDataSource === "Prometheus"
   )
 
   const fragmentDependencyCallback = useCallback(() => {
@@ -212,7 +214,6 @@ export default function Streaming() {
         }
       }
     }
-    return undefined
   }, [fragmentList, relationId])
 
   useEffect(() => {
@@ -223,21 +224,21 @@ export default function Streaming() {
         }
       }
     }
-    return () => { }
   }, [relationId, relationList, setRelationId])
 
-  // get back pressure rate without prometheus
-  const [backPressuresMetricsWithoutPromtheus, setBackPressuresMetrics] =
+  // get embedded back pressure rate from meta node
+  const [embeddedMetrics, setBackPressuresMetrics] =
     useState<BackPressuresMetrics>()
   const [previousBP, setPreviousBP] = useState<BackPressureInfo[]>([])
   const [currentBP, setCurrentBP] = useState<BackPressureInfo[]>([])
   const toast = useErrorToast()
 
   useEffect(() => {
-    if (backPressureDataSourceAlgo === "Embedded") {
+    if (backPressureDataSource === "Embedded") {
       const interval = setInterval(() => {
         const fetchNewBP = async () => {
-          const newBP = await getBackPressureWithoutPrometheus()
+          const newBP = await fetchEmbeddedBackPressure()
+          console.log(newBP)
           setPreviousBP(currentBP)
           setCurrentBP(newBP)
         }
@@ -246,7 +247,7 @@ export default function Streaming() {
       }, INTERVAL)
       return () => clearInterval(interval)
     }
-  }, [currentBP, backPressureDataSourceAlgo])
+  }, [currentBP, backPressureDataSource])
 
   useEffect(() => {
     if (currentBP !== null && previousBP !== null) {
@@ -323,24 +324,24 @@ export default function Streaming() {
   }
 
   const backPressures = useMemo(() => {
-    if (actorBackPressures || backPressuresMetricsWithoutPromtheus) {
+    if (promethusMetrics || embeddedMetrics) {
       let map = new Map()
 
       if (
-        backPressureDataSourceAlgo === "Embedded" &&
-        backPressuresMetricsWithoutPromtheus
+        backPressureDataSource === "Embedded" &&
+        embeddedMetrics
       ) {
-        for (const m of backPressuresMetricsWithoutPromtheus.outputBufferBlockingDuration) {
+        for (const m of embeddedMetrics.outputBufferBlockingDuration) {
           map.set(
             `${m.metric.fragmentId}_${m.metric.downstreamFragmentId}`,
             m.sample[0].value
           )
         }
       } else if (
-        backPressureDataSourceAlgo !== "Embedded" &&
-        actorBackPressures
+        backPressureDataSource !== "Embedded" &&
+        promethusMetrics
       ) {
-        for (const m of actorBackPressures.outputBufferBlockingDuration) {
+        for (const m of promethusMetrics.outputBufferBlockingDuration) {
           const value = average(m.sample) * 100
           map.set(
             `${m.metric.fragment_id}_${m.metric.downstream_fragment_id}`,
@@ -351,9 +352,9 @@ export default function Streaming() {
       return map
     }
   }, [
-    backPressureDataSourceAlgo,
-    actorBackPressures,
-    backPressuresMetricsWithoutPromtheus,
+    backPressureDataSource,
+    promethusMetrics,
+    embeddedMetrics,
   ])
 
   const retVal = (
@@ -427,9 +428,9 @@ export default function Streaming() {
           <FormControl>
             <FormLabel>Back Pressure Data Source</FormLabel>
             <Select
-              value={backPressureDataSourceAlgo}
+              value={backPressureDataSource}
               onChange={(event) =>
-                setBackPressureDataSourceAlgo(event.target.value)
+                setBackPressureDataSource(event.target.value)
               }
             >
               {backPressureDataSources.map((algo) => (

From ac0b063943d6d7cbc96ff995efbf57174e9c6710 Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Fri, 1 Mar 2024 17:25:28 +0800
Subject: [PATCH 03/13] rename

---
 dashboard/lib/api/metric.ts        |  9 +++++----
 dashboard/pages/fragment_graph.tsx | 14 +++++++++-----
 2 files changed, 14 insertions(+), 9 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index 647419672f7b6..0c180b8762a3e 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -21,8 +21,8 @@ export interface BackPressuresMetrics {
   outputBufferBlockingDuration: Metrics[]
 }
 
-// Get back pressure from meta node -> prometheus
-export async function getActorBackPressures() {
+// Get back pressure from Prometheus
+export async function fetchPrometheusBackPressure() {
   const res: BackPressuresMetrics = await api.get(
     "/metrics/fragment/prometheus_back_pressures"
   )
@@ -113,7 +113,8 @@ function convertToBackPressureMetrics(
 
 export function calculateBPRate(
   backPressureNew: BackPressureInfo[],
-  backPressureOld: BackPressureInfo[]
+  backPressureOld: BackPressureInfo[],
+  interval_ms: number
 ): BackPressuresMetrics {
   let mapNew = convertToMapAndAgg(backPressureNew)
   let mapOld = convertToMapAndAgg(backPressureOld)
@@ -123,7 +124,7 @@ export function calculateBPRate(
       result.set(
         key,
         // The *100 in end of the formular is to convert the BP rate to the value used in web UI drawing
-        ((value - (mapOld.get(key) || 0)) / ((INTERVAL / 1000) * 1000000000)) *
+        ((value - (mapOld.get(key) || 0)) / ((interval_ms / 1000) * 1000000000)) *
         100
       )
     } else {
diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index c47d00d21674b..aa823108ac562 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -43,7 +43,7 @@ import {
   BackPressuresMetrics,
   average,
   calculateBPRate,
-  getActorBackPressures,
+  fetchPrometheusBackPressure,
   fetchEmbeddedBackPressure,
 } from "../lib/api/metric"
 import { getFragments, getStreamingJobs } from "../lib/api/streaming"
@@ -195,7 +195,7 @@ export default function Streaming() {
     useState("Embedded")
 
   const { response: promethusMetrics } = useFetch(
-    getActorBackPressures,
+    fetchPrometheusBackPressure,
     INTERVAL,
     backPressureDataSource === "Prometheus"
   )
@@ -235,7 +235,8 @@ export default function Streaming() {
 
   useEffect(() => {
     if (backPressureDataSource === "Embedded") {
-      const interval = setInterval(() => {
+      const task = setInterval(() => {
+        console.log("Start polling embedded back-pressure metrics")
         const fetchNewBP = async () => {
           const newBP = await fetchEmbeddedBackPressure()
           console.log(newBP)
@@ -245,13 +246,16 @@ export default function Streaming() {
 
         fetchNewBP().catch(console.error)
       }, INTERVAL)
-      return () => clearInterval(interval)
+      return () => {
+        clearInterval(task);
+        console.log("Stop polling embedded back-pressure metrics")
+      }
     }
   }, [currentBP, backPressureDataSource])
 
   useEffect(() => {
     if (currentBP !== null && previousBP !== null) {
-      const metrics = calculateBPRate(currentBP, previousBP)
+      const metrics = calculateBPRate(currentBP, previousBP, INTERVAL)
       metrics.outputBufferBlockingDuration = sortBy(
         metrics.outputBufferBlockingDuration,
         (m) => (m.metric.fragmentId, m.metric.downstreamFragmentId)

From bb1159422f598ebb8e07a6369f1d36359a007bea Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Sat, 2 Mar 2024 10:36:50 +0800
Subject: [PATCH 04/13] refactor

---
 dashboard/pages/fragment_graph.tsx | 54 ++++++++++++++++--------------
 1 file changed, 29 insertions(+), 25 deletions(-)

diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index aa823108ac562..3bf8300516a2f 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -184,6 +184,13 @@ const backPressureDataSources: BackPressureDataSource[] = [
   "Prometheus",
 ]
 
+// The state of the embedded back pressure metrics.
+// The metrics from previous fetch are stored here to calculate the rate.
+interface EmbeddedBackPressureInfo {
+  previous: BackPressureInfo[]
+  current: BackPressureInfo[]
+}
+
 export default function Streaming() {
   const { response: relationList } = useFetch(getStreamingJobs)
   const { response: fragmentList } = useFetch(getFragments)
@@ -227,42 +234,33 @@ export default function Streaming() {
   }, [relationId, relationList, setRelationId])
 
   // get embedded back pressure rate from meta node
-  const [embeddedMetrics, setBackPressuresMetrics] =
-    useState<BackPressuresMetrics>()
-  const [previousBP, setPreviousBP] = useState<BackPressureInfo[]>([])
-  const [currentBP, setCurrentBP] = useState<BackPressureInfo[]>([])
+  const [embeddedBackPressureInfo, setEmbeddedBackPressureInfo] = useState<EmbeddedBackPressureInfo>()
   const toast = useErrorToast()
 
   useEffect(() => {
     if (backPressureDataSource === "Embedded") {
-      const task = setInterval(() => {
-        console.log("Start polling embedded back-pressure metrics")
+      console.log("Start polling embedded back-pressure metrics")
+      const interval = setInterval(() => {
         const fetchNewBP = async () => {
           const newBP = await fetchEmbeddedBackPressure()
           console.log(newBP)
-          setPreviousBP(currentBP)
-          setCurrentBP(newBP)
+          setEmbeddedBackPressureInfo((prev) => prev ? {
+            previous: prev.current,
+            current: newBP,
+          } : {
+            previous: newBP, // Use current value to show zero rate, but it's fine
+            current: newBP,
+          })
         }
 
         fetchNewBP().catch(console.error)
       }, INTERVAL)
       return () => {
-        clearInterval(task);
         console.log("Stop polling embedded back-pressure metrics")
+        clearInterval(interval);
       }
     }
-  }, [currentBP, backPressureDataSource])
-
-  useEffect(() => {
-    if (currentBP !== null && previousBP !== null) {
-      const metrics = calculateBPRate(currentBP, previousBP, INTERVAL)
-      metrics.outputBufferBlockingDuration = sortBy(
-        metrics.outputBufferBlockingDuration,
-        (m) => (m.metric.fragmentId, m.metric.downstreamFragmentId)
-      )
-      setBackPressuresMetrics(metrics)
-    }
-  }, [currentBP, previousBP])
+  }, [backPressureDataSource])
 
   const fragmentDependency = fragmentDependencyCallback()?.fragmentDep
   const fragmentDependencyDag = fragmentDependencyCallback()?.fragmentDepDag
@@ -328,14 +326,20 @@ export default function Streaming() {
   }
 
   const backPressures = useMemo(() => {
-    if (promethusMetrics || embeddedMetrics) {
+    if (promethusMetrics || embeddedBackPressureInfo) {
       let map = new Map()
 
       if (
         backPressureDataSource === "Embedded" &&
-        embeddedMetrics
+        embeddedBackPressureInfo
       ) {
-        for (const m of embeddedMetrics.outputBufferBlockingDuration) {
+        const metrics = calculateBPRate(embeddedBackPressureInfo.current, embeddedBackPressureInfo.previous, INTERVAL)
+        // why sorting?
+        metrics.outputBufferBlockingDuration = sortBy(
+          metrics.outputBufferBlockingDuration,
+          (m) => (m.metric.fragmentId, m.metric.downstreamFragmentId)
+        )
+        for (const m of metrics.outputBufferBlockingDuration) {
           map.set(
             `${m.metric.fragmentId}_${m.metric.downstreamFragmentId}`,
             m.sample[0].value
@@ -358,7 +362,7 @@ export default function Streaming() {
   }, [
     backPressureDataSource,
     promethusMetrics,
-    embeddedMetrics,
+    embeddedBackPressureInfo,
   ])
 
   const retVal = (

From bc779da8b974bba94769bffa9339ba5a1e92cb4a Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Sat, 2 Mar 2024 11:01:45 +0800
Subject: [PATCH 05/13] refactor2

---
 dashboard/pages/fragment_graph.tsx | 73 +++++++++++++++---------------
 1 file changed, 36 insertions(+), 37 deletions(-)

diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index 3bf8300516a2f..7281a16d6c1a9 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -197,15 +197,8 @@ export default function Streaming() {
 
   const [relationId, setRelationId] = useQueryState("id", parseAsInteger)
   const [selectedFragmentId, setSelectedFragmentId] = useState<number>()
-  // used to get the data source
-  const [backPressureDataSource, setBackPressureDataSource] =
-    useState("Embedded")
 
-  const { response: promethusMetrics } = useFetch(
-    fetchPrometheusBackPressure,
-    INTERVAL,
-    backPressureDataSource === "Prometheus"
-  )
+  const toast = useErrorToast()
 
   const fragmentDependencyCallback = useCallback(() => {
     if (fragmentList) {
@@ -233,35 +226,6 @@ export default function Streaming() {
     }
   }, [relationId, relationList, setRelationId])
 
-  // get embedded back pressure rate from meta node
-  const [embeddedBackPressureInfo, setEmbeddedBackPressureInfo] = useState<EmbeddedBackPressureInfo>()
-  const toast = useErrorToast()
-
-  useEffect(() => {
-    if (backPressureDataSource === "Embedded") {
-      console.log("Start polling embedded back-pressure metrics")
-      const interval = setInterval(() => {
-        const fetchNewBP = async () => {
-          const newBP = await fetchEmbeddedBackPressure()
-          console.log(newBP)
-          setEmbeddedBackPressureInfo((prev) => prev ? {
-            previous: prev.current,
-            current: newBP,
-          } : {
-            previous: newBP, // Use current value to show zero rate, but it's fine
-            current: newBP,
-          })
-        }
-
-        fetchNewBP().catch(console.error)
-      }, INTERVAL)
-      return () => {
-        console.log("Stop polling embedded back-pressure metrics")
-        clearInterval(interval);
-      }
-    }
-  }, [backPressureDataSource])
-
   const fragmentDependency = fragmentDependencyCallback()?.fragmentDep
   const fragmentDependencyDag = fragmentDependencyCallback()?.fragmentDepDag
   const fragments = fragmentDependencyCallback()?.fragments
@@ -325,6 +289,41 @@ export default function Streaming() {
     toast(new Error(`Actor ${searchActorIdInt} not found`))
   }
 
+  const [backPressureDataSource, setBackPressureDataSource] = useState("Embedded")
+
+  // Periodically fetch Prometheus back-pressure from Meta node
+  const { response: promethusMetrics } = useFetch(
+    fetchPrometheusBackPressure,
+    INTERVAL,
+    backPressureDataSource === "Prometheus"
+  )
+
+  // Periodically fetch embedded back-pressure from Meta node
+  // Didn't call `useFetch()` because the `setState` way is special.
+  const [embeddedBackPressureInfo, setEmbeddedBackPressureInfo] = useState<EmbeddedBackPressureInfo>()
+  useEffect(() => {
+    if (backPressureDataSource === "Embedded") {
+      const interval = setInterval(() => {
+        fetchEmbeddedBackPressure().then((newBP) => {
+          console.log(newBP)
+          setEmbeddedBackPressureInfo((prev) => prev ? {
+            previous: prev.current,
+            current: newBP,
+          } : {
+            previous: newBP, // Use current value to show zero rate, but it's fine
+            current: newBP,
+          })
+        }, (e) => {
+          console.error(e)
+          toast(e, "error")
+        })
+      }, INTERVAL)
+      return () => {
+        clearInterval(interval);
+      }
+    }
+  }, [backPressureDataSource])
+
   const backPressures = useMemo(() => {
     if (promethusMetrics || embeddedBackPressureInfo) {
       let map = new Map()

From ae23b7fd6181b9c080d4bad571e9285c2d9079a4 Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Sat, 2 Mar 2024 11:03:31 +0800
Subject: [PATCH 06/13] format

---
 dashboard/lib/api/metric.ts        |  5 ++-
 dashboard/pages/fragment_graph.tsx | 66 +++++++++++++++---------------
 2 files changed, 37 insertions(+), 34 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index 0c180b8762a3e..a0f4232a2ec4e 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -124,8 +124,9 @@ export function calculateBPRate(
       result.set(
         key,
         // The *100 in end of the formular is to convert the BP rate to the value used in web UI drawing
-        ((value - (mapOld.get(key) || 0)) / ((interval_ms / 1000) * 1000000000)) *
-        100
+        ((value - (mapOld.get(key) || 0)) /
+          ((interval_ms / 1000) * 1000000000)) *
+          100
       )
     } else {
       result.set(key, 0)
diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index 7281a16d6c1a9..471a9f062af75 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -40,11 +40,10 @@ import useErrorToast from "../hook/useErrorToast"
 import useFetch from "../lib/api/fetch"
 import {
   BackPressureInfo,
-  BackPressuresMetrics,
   average,
   calculateBPRate,
-  fetchPrometheusBackPressure,
   fetchEmbeddedBackPressure,
+  fetchPrometheusBackPressure,
 } from "../lib/api/metric"
 import { getFragments, getStreamingJobs } from "../lib/api/streaming"
 import { FragmentBox } from "../lib/layout"
@@ -289,7 +288,8 @@ export default function Streaming() {
     toast(new Error(`Actor ${searchActorIdInt} not found`))
   }
 
-  const [backPressureDataSource, setBackPressureDataSource] = useState("Embedded")
+  const [backPressureDataSource, setBackPressureDataSource] =
+    useState("Embedded")
 
   // Periodically fetch Prometheus back-pressure from Meta node
   const { response: promethusMetrics } = useFetch(
@@ -300,26 +300,34 @@ export default function Streaming() {
 
   // Periodically fetch embedded back-pressure from Meta node
   // Didn't call `useFetch()` because the `setState` way is special.
-  const [embeddedBackPressureInfo, setEmbeddedBackPressureInfo] = useState<EmbeddedBackPressureInfo>()
+  const [embeddedBackPressureInfo, setEmbeddedBackPressureInfo] =
+    useState<EmbeddedBackPressureInfo>()
   useEffect(() => {
     if (backPressureDataSource === "Embedded") {
       const interval = setInterval(() => {
-        fetchEmbeddedBackPressure().then((newBP) => {
-          console.log(newBP)
-          setEmbeddedBackPressureInfo((prev) => prev ? {
-            previous: prev.current,
-            current: newBP,
-          } : {
-            previous: newBP, // Use current value to show zero rate, but it's fine
-            current: newBP,
-          })
-        }, (e) => {
-          console.error(e)
-          toast(e, "error")
-        })
+        fetchEmbeddedBackPressure().then(
+          (newBP) => {
+            console.log(newBP)
+            setEmbeddedBackPressureInfo((prev) =>
+              prev
+                ? {
+                    previous: prev.current,
+                    current: newBP,
+                  }
+                : {
+                    previous: newBP, // Use current value to show zero rate, but it's fine
+                    current: newBP,
+                  }
+            )
+          },
+          (e) => {
+            console.error(e)
+            toast(e, "error")
+          }
+        )
       }, INTERVAL)
       return () => {
-        clearInterval(interval);
+        clearInterval(interval)
       }
     }
   }, [backPressureDataSource])
@@ -328,11 +336,12 @@ export default function Streaming() {
     if (promethusMetrics || embeddedBackPressureInfo) {
       let map = new Map()
 
-      if (
-        backPressureDataSource === "Embedded" &&
-        embeddedBackPressureInfo
-      ) {
-        const metrics = calculateBPRate(embeddedBackPressureInfo.current, embeddedBackPressureInfo.previous, INTERVAL)
+      if (backPressureDataSource === "Embedded" && embeddedBackPressureInfo) {
+        const metrics = calculateBPRate(
+          embeddedBackPressureInfo.current,
+          embeddedBackPressureInfo.previous,
+          INTERVAL
+        )
         // why sorting?
         metrics.outputBufferBlockingDuration = sortBy(
           metrics.outputBufferBlockingDuration,
@@ -344,10 +353,7 @@ export default function Streaming() {
             m.sample[0].value
           )
         }
-      } else if (
-        backPressureDataSource !== "Embedded" &&
-        promethusMetrics
-      ) {
+      } else if (backPressureDataSource !== "Embedded" && promethusMetrics) {
         for (const m of promethusMetrics.outputBufferBlockingDuration) {
           const value = average(m.sample) * 100
           map.set(
@@ -358,11 +364,7 @@ export default function Streaming() {
       }
       return map
     }
-  }, [
-    backPressureDataSource,
-    promethusMetrics,
-    embeddedBackPressureInfo,
-  ])
+  }, [backPressureDataSource, promethusMetrics, embeddedBackPressureInfo])
 
   const retVal = (
     <Flex p={3} height="calc(100vh - 20px)" flexDirection="column">

From ab8b87796a71f0b14453fd3f8e805043360b5d0a Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Sat, 2 Mar 2024 16:05:41 +0800
Subject: [PATCH 07/13] fix promethues metrics

---
 dashboard/pages/fragment_graph.tsx   | 29 +++++++++++++------------
 src/meta/src/dashboard/prometheus.rs | 32 ++++++++++++----------------
 2 files changed, 29 insertions(+), 32 deletions(-)

diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index 471a9f062af75..a3b5c6ced8395 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -311,13 +311,13 @@ export default function Streaming() {
             setEmbeddedBackPressureInfo((prev) =>
               prev
                 ? {
-                    previous: prev.current,
-                    current: newBP,
-                  }
+                  previous: prev.current,
+                  current: newBP,
+                }
                 : {
-                    previous: newBP, // Use current value to show zero rate, but it's fine
-                    current: newBP,
-                  }
+                  previous: newBP, // Use current value to show zero rate, but it's fine
+                  current: newBP,
+                }
             )
           },
           (e) => {
@@ -353,12 +353,11 @@ export default function Streaming() {
             m.sample[0].value
           )
         }
-      } else if (backPressureDataSource !== "Embedded" && promethusMetrics) {
+      } else if (backPressureDataSource === "Prometheus" && promethusMetrics) {
         for (const m of promethusMetrics.outputBufferBlockingDuration) {
-          const value = average(m.sample) * 100
           map.set(
             `${m.metric.fragment_id}_${m.metric.downstream_fragment_id}`,
-            value
+            m.sample[0].value * 100
           )
         }
       }
@@ -441,12 +440,14 @@ export default function Streaming() {
               onChange={(event) =>
                 setBackPressureDataSource(event.target.value)
               }
+              defaultValue="Embedded"
             >
-              {backPressureDataSources.map((algo) => (
-                <option value={algo} key={algo}>
-                  {algo}
-                </option>
-              ))}
+              <option value="Embedded" key="Embedded">
+                Embedded (5 secs)
+              </option>
+              <option value="Prometheus" key="Prometheus">
+                Prometheus (1 min)
+              </option>
             </Select>
           </FormControl>
           <Flex height="full" width="full" flexDirection="column">
diff --git a/src/meta/src/dashboard/prometheus.rs b/src/meta/src/dashboard/prometheus.rs
index 7baade4969749..22b41cb2f3506 100644
--- a/src/meta/src/dashboard/prometheus.rs
+++ b/src/meta/src/dashboard/prometheus.rs
@@ -17,7 +17,7 @@ use std::time::SystemTime;
 
 use anyhow::anyhow;
 use axum::{Extension, Json};
-use prometheus_http_query::response::{RangeVector, Sample};
+use prometheus_http_query::response::{InstantVector, RangeVector, Sample};
 use serde::Serialize;
 
 use super::handlers::{err, DashboardError};
@@ -53,6 +53,17 @@ impl From<&RangeVector> for PrometheusVector {
     }
 }
 
+// Note(eric): For backward compatibility, we store the `InstantVector` as a single sample,
+// instead of defining a new struct.
+impl From<&InstantVector> for PrometheusVector {
+    fn from(value: &InstantVector) -> Self {
+        PrometheusVector {
+            metric: value.metric().clone(),
+            sample: vec![PrometheusSample::from(value.sample())],
+        }
+    }
+}
+
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct ClusterMetrics {
@@ -134,27 +145,12 @@ pub async fn list_prometheus_fragment_back_pressure(
     Extension(srv): Extension<Service>,
 ) -> Result<Json<FragmentBackPressure>> {
     if let Some(ref client) = srv.prometheus_client {
-        let now = SystemTime::now();
         let back_pressure_query =
             format!("avg(rate(stream_actor_output_buffer_blocking_duration_ns{{{}}}[60s])) by (fragment_id, downstream_fragment_id) / 1000000000", srv.prometheus_selector);
-        let result = client
-            .query_range(
-                back_pressure_query,
-                now.duration_since(SystemTime::UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs() as i64
-                    - 1800,
-                now.duration_since(SystemTime::UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs() as i64,
-                15.0,
-            )
-            .get()
-            .await
-            .map_err(err)?;
+        let result = client.query(back_pressure_query).get().await.map_err(err)?;
         let back_pressure_data = result
             .data()
-            .as_matrix()
+            .as_vector()
             .unwrap()
             .iter()
             .map(PrometheusVector::from)

From 96de9ddaf6e04cb90478259183192efe76509cc7 Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Sat, 2 Mar 2024 16:15:09 +0800
Subject: [PATCH 08/13] clean

---
 dashboard/lib/api/metric.ts        |  8 --------
 dashboard/pages/fragment_graph.tsx | 13 ++++++-------
 2 files changed, 6 insertions(+), 15 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index a0f4232a2ec4e..4d49d941c99e7 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -181,14 +181,6 @@ export function p99(samples: MetricsSample[]) {
   return calculatePercentile(samples, 0.99)
 }
 
-export function average(samples: MetricsSample[]) {
-  let sum = 0
-  for (const sample of samples) {
-    sum += sample.value
-  }
-  return sum / samples.length
-}
-
 function isSet(value: any): boolean {
   return value !== null && value !== undefined
 }
diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index a3b5c6ced8395..c85dd1942ba9e 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -40,7 +40,6 @@ import useErrorToast from "../hook/useErrorToast"
 import useFetch from "../lib/api/fetch"
 import {
   BackPressureInfo,
-  average,
   calculateBPRate,
   fetchEmbeddedBackPressure,
   fetchPrometheusBackPressure,
@@ -311,13 +310,13 @@ export default function Streaming() {
             setEmbeddedBackPressureInfo((prev) =>
               prev
                 ? {
-                  previous: prev.current,
-                  current: newBP,
-                }
+                    previous: prev.current,
+                    current: newBP,
+                  }
                 : {
-                  previous: newBP, // Use current value to show zero rate, but it's fine
-                  current: newBP,
-                }
+                    previous: newBP, // Use current value to show zero rate, but it's fine
+                    current: newBP,
+                  }
             )
           },
           (e) => {

From 6c185e7959f3c04e7288599b8760827cf2669906 Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Mon, 4 Mar 2024 18:06:42 +0800
Subject: [PATCH 09/13] port some nice things from bugen's PR

---
 dashboard/pages/fragment_graph.tsx | 23 +++++++++++++++++------
 1 file changed, 17 insertions(+), 6 deletions(-)

diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index c85dd1942ba9e..d1d6b3fb1a23d 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -288,7 +288,7 @@ export default function Streaming() {
   }
 
   const [backPressureDataSource, setBackPressureDataSource] =
-    useState("Embedded")
+    useState<BackPressureDataSource>("Embedded")
 
   // Periodically fetch Prometheus back-pressure from Meta node
   const { response: promethusMetrics } = useFetch(
@@ -354,10 +354,19 @@ export default function Streaming() {
         }
       } else if (backPressureDataSource === "Prometheus" && promethusMetrics) {
         for (const m of promethusMetrics.outputBufferBlockingDuration) {
-          map.set(
-            `${m.metric.fragment_id}_${m.metric.downstream_fragment_id}`,
-            m.sample[0].value * 100
-          )
+          if (m.sample.length > 0) {
+            // Note: We issue an instant query to Prometheus to get the most recent value.
+            // So there should be only one sample here.
+            //
+            // Due to https://github.com/risingwavelabs/risingwave/issues/15280, it's still
+            // possible that an old version of meta service returns a range-query result.
+            // So we take the one with the latest timestamp here.
+            const value = _(m.sample).maxBy((s) => s.timestamp)!.value * 100
+            map.set(
+              `${m.metric.fragment_id}_${m.metric.downstream_fragment_id}`,
+              value
+            )
+          }
         }
       }
       return map
@@ -437,7 +446,9 @@ export default function Streaming() {
             <Select
               value={backPressureDataSource}
               onChange={(event) =>
-                setBackPressureDataSource(event.target.value)
+                setBackPressureDataSource(
+                  event.target.value as BackPressureDataSource
+                )
               }
               defaultValue="Embedded"
             >

From 5cee8fcf38af4a69af2853af501a945da2804b46 Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Mon, 4 Mar 2024 18:08:36 +0800
Subject: [PATCH 10/13] remove sorting

---
 dashboard/pages/fragment_graph.tsx | 17 ++++++-----------
 1 file changed, 6 insertions(+), 11 deletions(-)

diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index d1d6b3fb1a23d..d95a5f04f5eca 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -310,13 +310,13 @@ export default function Streaming() {
             setEmbeddedBackPressureInfo((prev) =>
               prev
                 ? {
-                    previous: prev.current,
-                    current: newBP,
-                  }
+                  previous: prev.current,
+                  current: newBP,
+                }
                 : {
-                    previous: newBP, // Use current value to show zero rate, but it's fine
-                    current: newBP,
-                  }
+                  previous: newBP, // Use current value to show zero rate, but it's fine
+                  current: newBP,
+                }
             )
           },
           (e) => {
@@ -341,11 +341,6 @@ export default function Streaming() {
           embeddedBackPressureInfo.previous,
           INTERVAL
         )
-        // why sorting?
-        metrics.outputBufferBlockingDuration = sortBy(
-          metrics.outputBufferBlockingDuration,
-          (m) => (m.metric.fragmentId, m.metric.downstreamFragmentId)
-        )
         for (const m of metrics.outputBufferBlockingDuration) {
           map.set(
             `${m.metric.fragmentId}_${m.metric.downstreamFragmentId}`,

From e9f128af2edc6e13d3d547f841b5ca70dad83027 Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Mon, 4 Mar 2024 18:10:14 +0800
Subject: [PATCH 11/13] minor: camel case

---
 dashboard/lib/api/metric.ts | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index 4d49d941c99e7..f9e78afac0033 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -114,7 +114,7 @@ function convertToBackPressureMetrics(
 export function calculateBPRate(
   backPressureNew: BackPressureInfo[],
   backPressureOld: BackPressureInfo[],
-  interval_ms: number
+  intervalMs: number
 ): BackPressuresMetrics {
   let mapNew = convertToMapAndAgg(backPressureNew)
   let mapOld = convertToMapAndAgg(backPressureOld)
@@ -125,8 +125,8 @@ export function calculateBPRate(
         key,
         // The *100 in end of the formular is to convert the BP rate to the value used in web UI drawing
         ((value - (mapOld.get(key) || 0)) /
-          ((interval_ms / 1000) * 1000000000)) *
-          100
+          ((intervalMs / 1000) * 1000000000)) *
+        100
       )
     } else {
       result.set(key, 0)

From 98238fbd6cf9c3aacb349c4119548a4a4737458f Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Tue, 5 Mar 2024 13:59:32 +0800
Subject: [PATCH 12/13] duplicates

---
 src/meta/src/dashboard/prometheus.rs | 11 +----------
 1 file changed, 1 insertion(+), 10 deletions(-)

diff --git a/src/meta/src/dashboard/prometheus.rs b/src/meta/src/dashboard/prometheus.rs
index ab6be56d05ca9..8b6ebe445de97 100644
--- a/src/meta/src/dashboard/prometheus.rs
+++ b/src/meta/src/dashboard/prometheus.rs
@@ -54,22 +54,13 @@ impl From<&RangeVector> for PrometheusVector {
     }
 }
 
-impl From<&InstantVector> for PrometheusVector {
-    fn from(value: &InstantVector) -> Self {
-        PrometheusVector {
-            metric: value.metric().clone(),
-            sample: vec![value.sample().into()],
-        }
-    }
-}
-
 // Note(eric): For backward compatibility, we store the `InstantVector` as a single sample,
 // instead of defining a new struct.
 impl From<&InstantVector> for PrometheusVector {
     fn from(value: &InstantVector) -> Self {
         PrometheusVector {
             metric: value.metric().clone(),
-            sample: vec![PrometheusSample::from(value.sample())],
+            sample: vec![value.sample().into()],
         }
     }
 }

From bb7d69d394285140da7fc7539cae64599387e13b Mon Sep 17 00:00:00 2001
From: Eric Fu <eric@singularity-data.com>
Date: Tue, 5 Mar 2024 14:03:34 +0800
Subject: [PATCH 13/13] format

---
 dashboard/lib/api/metric.ts        |  2 +-
 dashboard/pages/fragment_graph.tsx | 14 +++++++-------
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/dashboard/lib/api/metric.ts b/dashboard/lib/api/metric.ts
index f9e78afac0033..d40f92f405df1 100644
--- a/dashboard/lib/api/metric.ts
+++ b/dashboard/lib/api/metric.ts
@@ -126,7 +126,7 @@ export function calculateBPRate(
         // The *100 in end of the formular is to convert the BP rate to the value used in web UI drawing
         ((value - (mapOld.get(key) || 0)) /
           ((intervalMs / 1000) * 1000000000)) *
-        100
+          100
       )
     } else {
       result.set(key, 0)
diff --git a/dashboard/pages/fragment_graph.tsx b/dashboard/pages/fragment_graph.tsx
index d95a5f04f5eca..e72ca7ce3e0ce 100644
--- a/dashboard/pages/fragment_graph.tsx
+++ b/dashboard/pages/fragment_graph.tsx
@@ -29,7 +29,7 @@ import {
 } from "@chakra-ui/react"
 import * as d3 from "d3"
 import { dagStratify } from "d3-dag"
-import _, { sortBy } from "lodash"
+import _ from "lodash"
 import Head from "next/head"
 import { parseAsInteger, useQueryState } from "nuqs"
 import { Fragment, useCallback, useEffect, useMemo, useState } from "react"
@@ -310,13 +310,13 @@ export default function Streaming() {
             setEmbeddedBackPressureInfo((prev) =>
               prev
                 ? {
-                  previous: prev.current,
-                  current: newBP,
-                }
+                    previous: prev.current,
+                    current: newBP,
+                  }
                 : {
-                  previous: newBP, // Use current value to show zero rate, but it's fine
-                  current: newBP,
-                }
+                    previous: newBP, // Use current value to show zero rate, but it's fine
+                    current: newBP,
+                  }
             )
           },
           (e) => {