feat: update event grid to support live streaming of data

Author: starpit

plugins/plugin-client-default/notebooks/dashboard.md

@@ -80,5 +80,5 @@ layout:
     execute: now
     outputOnly: true
     ---
-    chart progress "${LOGDIR}"
+    chart events "${LOGDIR}"
     ```

plugins/plugin-codeflare/src/components/Grid.tsx

@@ -17,18 +17,18 @@
 import React from "react"
 import { Tooltip } from "@kui-shell/plugin-client-common"
 
-import Event from "../controller/events/Event"
+import { GenericEvent } from "../controller/events/Event"
 
 import "@kui-shell/plugin-client-common/web/scss/components/Table/_index.scss"
 import "@kui-shell/plugin-client-common/web/scss/components/Table/Grid/_index.scss"
 import "../../web/scss/components/Dashboard/Grid.scss"
 
 interface Props {
-  events: Event<string, unknown>[]
+  events: GenericEvent[]
 }
 
 export default class Grid extends React.PureComponent<Props> {
-  private tooltipContent(event: Event<string, unknown>) {
+  private tooltipContent(event: GenericEvent) {
     const title = event.name
     const subtitle = event.subtitle || event.type
     const status = event.state
@@ -42,7 +42,7 @@ ${status ? "Status: " + status : ""}
 \`${showMoreDetail}\``
   }
 
-  private readonly cell = (event: Event<string, unknown>, idx: number) => {
+  private readonly cell = (event: GenericEvent, idx: number) => {
     return (
       <Tooltip key={idx} markdown={this.tooltipContent(event)}>
         <span className="kui--grid-cell" data-tag="badge" data-type={event.type} data-state={event.state}>

plugins/plugin-codeflare/src/controller/events/Event.ts

@@ -15,6 +15,7 @@
  */
 
 export type State = "InProgress" | "Done" | "Error"
+
 type Event<T extends string, Detail> = Detail & {
   name: string
   subtitle?: string
@@ -25,4 +26,6 @@ type Event<T extends string, Detail> = Detail & {
   hidden?: boolean
 }
 
+export type GenericEvent = Event<string, unknown>
+
 export default Event

plugins/plugin-codeflare/src/controller/events/Events.tsx

@@ -19,22 +19,184 @@ import { join } from "path"
 import stripAnsi from "strip-ansi"
 import { Arguments } from "@kui-shell/core"
 
-import kubeEvents from "./kube"
-import torchEvents from "./torch"
+import parseKubeEvents, { collateEvent as collateKubeEvent, KubeEvent } from "./kube"
+import parseTorchEvents, { collateEvent as collateTorchEvent, TorchEvent } from "./torch"
 
 import { expand } from "../../lib/util"
 import Grid from "../../components/Grid"
 
+interface EventState {
+  kubeEvents: KubeEvent[]
+  torchEvents: TorchEvent[]
+}
+
+type State = EventState & {
+  nKubeEvents: number
+  nTorchEvents: number
+  catastrophicError?: Error
+}
+
+type Props = EventState & {
+  /** Follow kube events? */
+  onKube?(eventType: "data", cb: (data: any) => void): void
+
+  /** Follow torch events? */
+  onTorch?(eventType: "data", cb: (data: any) => void): void
+
+  /** Stop watching? */
+  unwatch?(): void
+}
+
+class Events extends React.PureComponent<Props, State> {
+  public constructor(props: Props) {
+    super(props)
+
+    const kubeEvents = props.kubeEvents || []
+    const torchEvents = props.torchEvents || []
+    this.state = {
+      kubeEvents,
+      torchEvents,
+      nKubeEvents: kubeEvents.length,
+      nTorchEvents: torchEvents.length,
+    }
+
+    // reduce any initial flood of events
+    let queueFlushHysteresis = 300
+    setTimeout(() => (queueFlushHysteresis = 0), 5000)
+
+    if (props.onKube) {
+      let queue: string[] = []
+      let flushTO: ReturnType<typeof setTimeout>
+
+      props.onKube("data", (line) => {
+        if (typeof line === "string") {
+          queue.push(stripAnsi(line))
+
+          if (flushTO) {
+            clearTimeout(flushTO)
+          }
+
+          flushTO = setTimeout(() => {
+            const toBeProcessed = queue
+            queue = []
+            this.setState((curState) => {
+              toBeProcessed.forEach((line) => collateKubeEvent(curState.kubeEvents, line))
+              return {
+                nKubeEvents: curState.kubeEvents.length,
+              }
+            })
+          }, queueFlushHysteresis)
+        }
+      })
+    }
+    if (props.onTorch) {
+      let queue: string[] = []
+      let flushTO: ReturnType<typeof setTimeout>
+
+      props.onTorch("data", (line) => {
+        if (typeof line === "string") {
+          queue.push(stripAnsi(line))
+
+          if (flushTO) {
+            clearTimeout(flushTO)
+          }
+
+          flushTO = setTimeout(() => {
+            const toBeProcessed = queue
+            queue = []
+            this.setState((curState) => {
+              toBeProcessed.forEach((line) => collateTorchEvent(curState.torchEvents, line))
+              return {
+                nTorchEvents: curState.torchEvents.length,
+              }
+            })
+          }, queueFlushHysteresis)
+        }
+      })
+    }
+  }
+
+  public static getDerivedStateFromError(error: Error) {
+    return { catastrophicError: error }
+  }
+
+  public componentDidCatch(error: Error, errorInfo: React.ErrorInfo) {
+    console.error("catastrophic error in Scalar", error, errorInfo)
+  }
+
+  public componentWillUnmount() {
+    if (this.props.unwatch) {
+      this.props.unwatch()
+    }
+  }
+
+  private get events() {
+    return [...this.state.kubeEvents, ...this.state.torchEvents]
+      .filter((_) => !_.hidden)
+      .sort((a, b) => a.timestamp - b.timestamp)
+  }
+
+  public render() {
+    if (this.state.catastrophicError) {
+      return "Internal Error"
+    } else {
+      return <Grid events={this.events} />
+    }
+  }
+}
+
 async function eventsUI(filepath: string, REPL: Arguments["REPL"]) {
-  const [kube, logs] = await Promise.all([
-    REPL.qexec<string>(`vfs fslice ${join(expand(filepath), "events/kubernetes.txt")} 0`).then(stripAnsi),
-    REPL.qexec<string>(`vfs fslice ${join(expand(filepath), "logs/job.txt")} 0`).then(stripAnsi),
-  ])
-
-  const events = [...kubeEvents(kube), ...torchEvents(logs)]
-    .filter((_) => !_.hidden)
-    .sort((a, b) => a.timestamp - b.timestamp)
-  return <Grid events={events} />
+  const jobFilepath = join(expand(filepath), "logs/job.txt")
+  const kubeFilepath = join(expand(filepath), "events/kubernetes.txt")
+
+  if (process.env.FOLLOW) {
+    const [TailFile, split2] = await Promise.all([
+      import("@logdna/tail-file").then((_) => _.default),
+      import("split2").then((_) => _.default),
+    ])
+
+    const kubeTail = new TailFile(kubeFilepath, {
+      startPos: 0,
+      pollFileIntervalMs: 500,
+    })
+    kubeTail.on("tail_error", (err) => {
+      console.error(err)
+    })
+
+    const jobTail = new TailFile(jobFilepath, {
+      startPos: 0,
+      pollFileIntervalMs: 500,
+    })
+    jobTail.on("tail_error", (err) => {
+      console.error(err)
+    })
+
+    kubeTail.start()
+    jobTail.start()
+
+    const kubeSplitter = kubeTail.pipe(split2())
+    const torchSplitter = jobTail.pipe(split2())
+
+    return (
+      <Events
+        kubeEvents={[]}
+        torchEvents={[]}
+        onKube={kubeSplitter.on.bind(kubeSplitter)}
+        onTorch={torchSplitter.on.bind(torchSplitter)}
+        unwatch={() => {
+          kubeTail.quit()
+          jobTail.quit()
+        }}
+      />
+    )
+  } else {
+    const [kube, torch] = await Promise.all([
+      REPL.qexec<string>(`vfs fslice ${kubeFilepath} 0`).then(stripAnsi).then(parseKubeEvents),
+      REPL.qexec<string>(`vfs fslice ${jobFilepath} 0`).then(stripAnsi).then(parseTorchEvents),
+    ])
+
+    return <Events kubeEvents={kube} torchEvents={torch} />
+  }
 }
 
 export default async function eventsCmd(args: Arguments) {

plugins/plugin-codeflare/src/controller/events/index.ts

@@ -18,5 +18,5 @@ import { Registrar } from "@kui-shell/core"
 
 /** Register Kui Commands for rendering dashboard event UIs */
 export default function registerEventCommands(registrar: Registrar) {
-  registrar.listen("/chart/progress", (args) => import("./Events").then((_) => _.default(args)), { needsUI: true })
+  registrar.listen("/chart/events", (args) => import("./Events").then((_) => _.default(args)), { needsUI: true })
 }

plugins/plugin-codeflare/src/controller/events/kube.ts

@@ -17,7 +17,7 @@
 import Event from "./Event"
 
 type EventType = "Pulling" | "Pulled"
-type KubeEvent = Event<EventType, { node: string }>
+export type KubeEvent = Event<EventType, { node: string }>
 
 function findPrevious(M: KubeEvent[], node: KubeEvent["node"], type: EventType) {
   for (let idx = M.length - 1; idx >= 0; idx--) {
@@ -28,7 +28,7 @@ function findPrevious(M: KubeEvent[], node: KubeEvent["node"], type: EventType)
   }
 }
 
-function collateEvent(M: KubeEvent[], line: string) {
+export function collateEvent(M: KubeEvent[], line: string): KubeEvent[] {
   const pullMatch = line.match(/(Pulling|Pulled)\s+(\S+)\s+(.+)$/)
   if (pullMatch) {
     const type = pullMatch[1] as EventType

plugins/plugin-codeflare/src/controller/events/torch.ts

@@ -18,7 +18,7 @@ import Event from "./Event"
 
 type EventType = "Epoch" | "Iteration" | "Marker"
 type Detail = { epoch: number; step: number; nSteps: number; ip: string }
-type TorchEvent = Event<EventType, Detail>
+export type TorchEvent = Event<EventType, Detail>
 
 function findPrevious(M: TorchEvent[], ip: TorchEvent["ip"], type: EventType) {
   for (let idx = M.length - 1; idx >= 0; idx--) {
@@ -34,7 +34,7 @@ function findEpoch(M: TorchEvent[], ip: TorchEvent["ip"]) {
   return evt ? evt.step : -1
 }
 
-function collateEvent(M: TorchEvent[], line: string) {
+export function collateEvent(M: TorchEvent[], line: string) {
   const startMatch = line.match(/ip=([\d.]+)\)\s+(\d+\/\d+\/\d+\s+\d+:\d+:\d+)\s+.+\*\*\*\*\* Running training/)
   if (startMatch) {
     const ip = startMatch[1]

plugins/plugin-codeflare/web/scss/components/Dashboard/Grid.scss

@@ -77,6 +77,12 @@ $fullWidth: 1em; /* $large * ($unit + $rgap) - $rgap */
   grid-template-columns: repeat(auto-fill, $unit) !important;
   grid-auto-rows: $fullWidth;
 
+  /**
+   * Cosmetic: to keep the cursor from switching back and forth from
+   * pointer to default as you sweep your mouse along the events.
+   */
+  cursor: pointer;
+
   @include CFCell {
     padding-left: 1px;
     grid-column: span $large;