fix: use a priority heap to choose most important events to display

e.g. if we are limited in vertical space, prefer to show pods still Pulling images over pods that have announced they are done and Pulled
also this improves the display of duplicate lines, rather just updating the timestamp

Author: starpit

package-lock.json

@@ -23,12 +23,14 @@
     "access": "public"
   },
   "devDependencies": {
+    "@types/heap": "^0.2.31",
     "@types/split2": "^3.2.1",
     "react-devtools-core": "^4.27.4"
   },
   "dependencies": {
     "@logdna/tail-file": "^3.0.1",
     "chokidar": "^3.5.3",
+    "heap": "^0.2.7",
     "ink": "^3.2.0",
     "madwizard": "^8.0.3",
     "pretty-ms": "^7.0.1",

plugins/plugin-codeflare-dashboard/package.json

@@ -86,7 +86,7 @@ export default class Dashboard extends React.PureComponent<Props, State> {
     this.setState((curState) => ({
       firstUpdate: (curState && curState.firstUpdate) || Date.now(), // TODO pull from the events
       lastUpdate: Date.now(), // TODO pull from the events
-      lines: !model.lines || model.lines.N === 0 ? curState?.lines : model.lines,
+      lines: !model.lines || model.lines.length === 0 ? curState?.lines : model.lines,
       workers: !curState?.workers
         ? [model.workers]
         : [...curState.workers.slice(0, gridIdx), model.workers, ...curState.workers.slice(gridIdx + 1)],
@@ -168,26 +168,26 @@ export default class Dashboard extends React.PureComponent<Props, State> {
     return prettyMillis(millis, { compact: true }) + " ago"
   }
 
-  private agos(timestamp: string) {
-    return this.ago(Date.now() - new Date(timestamp).getTime())
+  private agos(millis: number) {
+    return this.ago(Date.now() - millis)
   }
 
   private footer() {
     if (!this.lines) {
       return <React.Fragment />
     } else {
-      const rows = []
-      for (let idx = this.lines.idx, n = 0; n < this.lines.N; idx = (idx + 1) % this.lines.lines.length, n++) {
-        const line = this.lines.lines[idx]
-          .replace(/(\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\dZ)/, (_, timestamp) => this.agos(timestamp))
-          .replace(/pod\/torchx-\S+ /, "") // worker name in torchx
-          .replace(/pod\/ray-(head|worker)-\S+ /, "") // worker name in ray
-          .replace(/\* /, "") // wildcard worker name (codeflare)
+      const rows = this.lines.map(({ line, timestamp }) => {
+        const txt = line.replace("{timestamp}", () => this.agos(timestamp))
+        //          .replace(/pod\/torchx-\S+ /, "") // worker name in torchx
+        //          .replace(/pod\/ray-(head|worker)-\S+ /, "") // worker name in ray
+        //          .replace(/\* /, "") // wildcard worker name (codeflare)
+        //          .replace(/\x1b\x5B\[2J/g, "")
+        // TODO timestamp
 
         // [2J is part of clear screen; we don't want those to flow through
         // eslint-disable-next-line no-control-regex
-        rows.push(<Text key={line + "-" + n}>{line.replace(/\x1b\x5B\[2J/g, "")}</Text>)
-      }
+        return <Text key={txt}>{txt}</Text>
+      })
       return (
         <Box marginTop={1} flexDirection="column">
           {rows}

plugins/plugin-codeflare-dashboard/src/components/Dashboard/index.tsx

@@ -39,7 +39,7 @@ export type UpdatePayload = {
   workers: Worker[]
 
   /** Lines of raw output to be displayed */
-  lines?: { lines: string[]; idx: number; N: number }
+  lines?: { line: string; timestamp: number }[]
 }
 
 /** Callback from controller when it has updated data */

plugins/plugin-codeflare-dashboard/src/components/Dashboard/types.ts

@@ -14,13 +14,16 @@
  * limitations under the License.
  */
 
+import Heap from "heap"
 import stripAnsi from "strip-ansi"
 import type { TextProps } from "ink"
 
 import type { Tail } from "../tailf.js"
 import type { OnData, Worker } from "../../../components/Dashboard/types.js"
 
-import { WorkerState, stateFor } from "./states.js"
+import { WorkerState, rankFor, stateFor } from "./states.js"
+
+type Line = { line: string; stateRank: number; timestamp: number }
 
 /**
  * Maintain a model of live data from a given set of file streams
@@ -32,16 +35,21 @@ export default class Live {
   private readonly workers: Record<string, Worker> = {}
 
   /** Number of lines of output to retain. TODO this depends on height of terminal? */
-  private static readonly nLines = 18
-
-  /** Model of the last `Live.nLines` lines of output */
-  private readonly lines: string[] = Array(Live.nLines).fill("")
+  private static readonly MAX_HEAP = 1000
 
-  /** Current insertion index into `this.lines` */
-  private linesInsertionIdx = 0
+  /** Model of the lines of output */
+  private readonly lines = new Heap<Line>((a, b) => {
+    if (a.line === b.line) {
+      return 0
+    }
 
-  /** Current number of valid lines in `this.lines` */
-  private linesCount = 0
+    const stateDiff = a.stateRank - b.stateRank
+    if (stateDiff !== 0) {
+      return stateDiff
+    } else {
+      return a.timestamp - b.timestamp
+    }
+  })
 
   public constructor(private readonly tails: Promise<Tail>[], cb: OnData, styleOf: Record<WorkerState, TextProps>) {
     tails.map((tailf) => {
@@ -60,7 +68,7 @@ export default class Live {
 
             if (!name || !timestamp) {
               // console.error("Bad status record", line)
-              this.pushLineAndPublish(data, cb)
+              // this.pushLineAndPublish(data, metric, timestamp, cb)
               return
             } else if (!metric) {
               // ignoring this line
@@ -93,7 +101,7 @@ export default class Live {
 
                 // inform the UI that we have updates
                 cb({
-                  lines: this.pushLine(data),
+                  lines: this.pushLine(data, metric, timestamp),
                   workers: Object.values(this.workers),
                 })
               }
@@ -112,40 +120,45 @@ export default class Live {
     })
   }
 
+  private readonly lookup: Record<string, Line> = {}
   /** Add `line` to our circular buffer `this.lines` */
-  private pushLine(line: string) {
-    if (this.lines.includes(line)) {
-      // duplicate line
-      // the oldest is the one we are about to overwrite
-      let oldestIdx = this.linesInsertionIdx
-      if (this.lines[oldestIdx].length === 0) {
-        do {
-          oldestIdx = (oldestIdx + 1) % Live.nLines
-        } while (this.lines[oldestIdx].length === 0)
-      }
-      return { lines: this.lines, idx: oldestIdx, N: this.linesCount }
+  private pushLine(line: string, metric: WorkerState, timestamp: number) {
+    const key = line
+      .replace(/\s*(\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\dZ)\s*/, "{timestamp}")
+      .replace(/pod\/torchx-\S+ /, "") // worker name in torchx
+      .replace(/pod\/ray-(head|worker)-\S+ /, "") // worker name in ray
+      .replace(/\* /, "") // wildcard worker name (codeflare)
+      .replace(/\x1b\x5B\[2J/g, "") // eslint-disable-line no-control-regex
+    // ^^^ [2J is part of clear screen; we don't want those to flow through
+
+    const rec = {
+      timestamp,
+      stateRank: rankFor[metric],
+      line: key,
+    }
+
+    const already = this.lookup[rec.line]
+    if (already) {
+      already.timestamp = timestamp
+      this.lines.updateItem(already)
     } else {
-      const idx = this.linesInsertionIdx
-      this.linesInsertionIdx = (this.linesInsertionIdx + 1) % Live.nLines
-
-      this.lines[idx] = line
-      this.linesCount = Math.min(this.lines.length, this.linesCount + 1)
-
-      // the oldest is the one we are about to overwrite
-      let oldestIdx = this.linesInsertionIdx
-      if (this.lines[oldestIdx].length === 0) {
-        do {
-          oldestIdx = (oldestIdx + 1) % Live.nLines
-        } while (this.lines[oldestIdx].length === 0)
+      this.lookup[rec.line] = rec
+      if (this.lines.size() >= Live.MAX_HEAP) {
+        this.lines.replace(rec)
+      } else {
+        this.lines.push(rec)
       }
-
-      return { lines: this.lines, idx: oldestIdx, N: this.linesCount }
     }
+
+    return this.lines
+      .toArray()
+      .slice(0, 18)
+      .sort((a, b) => a.timestamp - b.timestamp)
   }
 
   /** `pushLine` and then pass the updated model to `cb` */
-  private pushLineAndPublish(line: string, cb: OnData) {
-    cb({ lines: this.pushLine(line), workers: Object.values(this.workers) })
+  private pushLineAndPublish(line: string, metric: WorkerState, timestamp: number, cb: OnData) {
+    cb({ lines: this.pushLine(line, metric, timestamp), workers: Object.values(this.workers) })
   }
 
   private asMillisSinceEpoch(timestamp: string) {

plugins/plugin-codeflare-dashboard/src/controller/dashboard/status/Live.ts

@@ -19,6 +19,16 @@ export const states = ["Queued", "Provisioning", "Initializing", "Running", "Suc
 /** Type declaration for quantized utilization states */
 export type WorkerState = (typeof states)[number]
 
+/** Lower means more important */
+export const rankFor: Record<WorkerState, number> = {
+  Queued: 3,
+  Provisioning: 1,
+  Initializing: 2,
+  Running: 1,
+  Success: 0,
+  Failed: 0,
+}
+
 export const stateFor: Record<string, WorkerState> = {
   Pending: "Queued",
   Queued: "Queued",
@@ -32,9 +42,9 @@ export const stateFor: Record<string, WorkerState> = {
 
   Unhealthy: "Provisioning", // service not yet ready... Initializing?
 
-  Initializing: "Initializing",
-  Installing: "Initializing",
-  Pulling: "Initializing",
+  Initializing: "Provisioining",
+  Installing: "Provisioining",
+  Pulling: "Provisioining",
   Pulled: "Initializing",
 
   Started: "Initializing",

plugins/plugin-codeflare-dashboard/src/controller/dashboard/status/states.ts

@@ -48,10 +48,14 @@ export async function getJobEnv(profile: string, jobId: string): Promise<Record<
 }
 
 export async function numGpus(profile: string, jobId: string): Promise<number> {
-  const env = await getJobEnv(profile, jobId)
+  try {
+    const env = await getJobEnv(profile, jobId)
 
-  const raw = env["NUM_GPUS"]
-  return typeof raw === "number" ? raw : typeof raw === "string" ? parseInt(raw, 10) : 0
+    const raw = env["NUM_GPUS"]
+    return typeof raw === "number" ? raw : typeof raw === "string" ? parseInt(raw, 10) : 0
+  } catch (err) {
+    return 0
+  }
 }
 
 export async function usesGpus(profile: string, jobId: string): Promise<boolean> {