feat: Extend kubectl optimizations to table watching

Fixes #6449

Author: starpit

packages/core/src/webapp/models/table.ts

@@ -135,6 +135,9 @@ export class Table<RowType extends Row = Row> {
   /** Markdown cells? */
   markdown?: boolean
 
+  /** This field helps with watching/paginating */
+  resourceVersion?: number | string
+
   /** Default presentation? */
   defaultPresentation?: PresentationStyle
 

plugins/plugin-bash-like/src/pty/server.ts

@@ -30,7 +30,7 @@ import { IncomingMessage } from 'http'
 import { Channel } from './channel'
 import { StdioChannelKuiSide } from './stdio-channel'
 
-import { CodedError, ExecOptions, Registrar, expandHomeDir } from '@kui-shell/core'
+import { Abortable, CodedError, ExecOptions, FlowControllable, Registrar, expandHomeDir } from '@kui-shell/core'
 
 const debug = Debug('plugins/bash-like/pty/server')
 
@@ -237,6 +237,9 @@ export const onConnection = (exitNow: ExitHandler, uid?: number, gid?: number) =
   // @see https://github.com/microsoft/TypeScript/issues/22445
   const shells: Record<string, Promise<import('node-pty-prebuilt-multiarch').IPty>> = {}
 
+  /** Active streaming jobs */
+  const jobs: Record<string, Abortable & FlowControllable> = {}
+
   // For all websocket data send it to the shell
   ws.on('message', async (data: string) => {
     try {
@@ -251,6 +254,9 @@ export const onConnection = (exitNow: ExitHandler, uid?: number, gid?: number) =
         rows?: number
         cols?: number
 
+        /** for type:request, execute a command, and stream back the output; the default behavior is request/respo;nse */
+        stream?: boolean
+
         uuid?: string // for request-response
         execOptions?: ExecOptions
       } = JSON.parse(data)
@@ -262,6 +268,11 @@ export const onConnection = (exitNow: ExitHandler, uid?: number, gid?: number) =
             const RESUME = '\x11' // this is XON
             shell.write(RESUME)
           }
+
+          const job = msg.uuid && jobs[msg.uuid]
+          if (job) {
+            job.xon()
+          }
           break
         }
 
@@ -271,6 +282,11 @@ export const onConnection = (exitNow: ExitHandler, uid?: number, gid?: number) =
             const PAUSE = '\x13' // this is XOFF
             shell.write(PAUSE)
           }
+
+          const job = msg.uuid && jobs[msg.uuid]
+          if (job) {
+            job.xoff()
+          }
           break
         }
 
@@ -280,6 +296,12 @@ export const onConnection = (exitNow: ExitHandler, uid?: number, gid?: number) =
             shell.kill(msg.signal || 'SIGHUP')
             return exitNow(msg.exitCode || 0)
           }
+
+          const job = msg.uuid && jobs[msg.uuid]
+          debug(`kill requested. hasShell=${shell !== undefined} hasJob=${job !== undefined}`)
+          if (job) {
+            job.abort()
+          }
           break
         }
 
@@ -297,8 +319,25 @@ export const onConnection = (exitNow: ExitHandler, uid?: number, gid?: number) =
             // ws.send(`___kui_exit___ ${msg.uuid}`)
           }
 
+          const execOptions = Object.assign({}, msg.execOptions, { rethrowErrors: true })
+          if (msg.stream) {
+            // then we will stream back the output; otherwise, we will use a request/response style of execution
+            debug('initializing streaming exec')
+            execOptions.onInit = (job: Abortable & FlowControllable) => {
+              jobs[msg.uuid] = job
+              return chunk =>
+                ws.send(
+                  JSON.stringify({
+                    type: 'chunk',
+                    uuid: msg.uuid,
+                    chunk
+                  })
+                )
+            }
+          }
+
           try {
-            const response = await exec(msg.cmdline, Object.assign({}, msg.execOptions, { rethrowErrors: true }))
+            const response = await exec(msg.cmdline, execOptions)
             debug('got response')
             terminate(
               JSON.stringify({

plugins/plugin-client-common/web/scss/components/Table/Events.scss

@@ -71,7 +71,7 @@ $inter-message-spacing: 0.1875rem;
 
   a {
     color: var(--color-base0F);
-    min-width: 4rem;
+    font-size: 0.75rem;
     display: inline-flex;
   }
 }

plugins/plugin-kubectl/src/controller/client/direct/errors.ts

@@ -16,10 +16,10 @@
 
 import { CodedError, isCodedError, REPL } from '@kui-shell/core'
 
+import { URLFormatter } from './get'
 import { Status, isStatus } from '../../../lib/model/resource'
 import { fetchFile, FetchedFile, isReturnedError } from '../../../lib/util/fetch-file'
 
-type URL = (includeKind: boolean) => string
 type WithErrors = { errors: CodedError[]; ok: (string | Buffer | object)[] }
 
 /** See if the given error message is a Kubernetes Status object */
@@ -38,7 +38,7 @@ function tryParseAsStatus(message: string): string | Status {
 
 export default async function handleErrors(
   responses: FetchedFile[],
-  formatUrl: URL,
+  formatUrl: URLFormatter,
   kind: string,
   repl: REPL
 ): Promise<WithErrors> {

plugins/plugin-kubectl/src/controller/client/direct/get.ts

@@ -16,6 +16,7 @@
 
 import { Arguments, CodedError, ExecType } from '@kui-shell/core'
 
+import makeWatchable from './watch'
 import { Explained } from '../../kubectl/explain'
 import { fetchFile } from '../../../lib/util/fetch-file'
 import { toKuiTable, withNotFound } from '../../../lib/view/formatTable'
@@ -33,12 +34,14 @@ import {
 import handleErrors from './errors'
 import { isStatus, KubeItems, MetaTable } from '../../../lib/model/resource'
 
-export default async function getDirect(args: Arguments<KubeOptions>, _kind: Promise<Explained>) {
-  const format = formatOf(args)
-  const kindIdx = args.argvNoOptions.indexOf('get') + 1
-  const { kind, version, isClusterScoped } = _kind
-    ? await _kind
-    : { kind: undefined, version: undefined, isClusterScoped: false }
+export type URLFormatter = (includeKind?: boolean, includeQueries?: boolean, name?: string) => string
+
+export const headersForTableRequest = { accept: 'application/json;as=Table;g=meta.k8s.io;v=v1' }
+
+export async function urlFormatterFor(
+  args: Arguments<KubeOptions>,
+  { kind, version, isClusterScoped }: Explained
+): Promise<URLFormatter> {
   const namespace = await getNamespace(args)
 
   const kindOnPath = `/${encodeURIComponent(kind.toLowerCase() + (/s$/.test(kind) ? '' : 's'))}`
@@ -76,24 +79,27 @@ export default async function getDirect(args: Arguments<KubeOptions>, _kind: Pro
   }
 
   // format a url
-  const formatUrl = (includeKind = false, includeQueries = false, name?: string) =>
+  return (includeKind = false, includeQueries = false, name?: string) =>
     `kubernetes:///${apiOnPath}${namespaceOnPath}${!includeKind ? '' : kindOnPath}${
       !name ? '' : `/${encodeURIComponent(name)}`
     }${!includeQueries || queries.length === 0 ? '' : '?' + queries.join('&')}`
+}
+
+export default async function getDirect(args: Arguments<KubeOptions>, _kind: Promise<Explained>) {
+  const explainedKind = _kind ? await _kind : { kind: undefined, version: undefined, isClusterScoped: false }
+  const { kind } = explainedKind
+  const formatUrl = await urlFormatterFor(args, explainedKind)
 
+  const format = formatOf(args)
+  const kindIdx = args.argvNoOptions.indexOf('get') + 1
   const names = args.argvNoOptions.slice(kindIdx + 1)
   const urls = names.length === 0 ? formatUrl(true, true) : names.map(formatUrl.bind(undefined, true, true)).join(',')
 
-  if (isTableRequest(args) && !isWatchRequest(args)) {
+  if (isTableRequest(args)) {
     const fmt = format || 'default'
     if (fmt === 'wide' || fmt === 'default') {
       // first, fetch the data
-      const responses = await fetchFile(
-        args.REPL,
-        urls,
-        { accept: 'application/json;as=Table;g=meta.k8s.io;v=v1' },
-        true
-      )
+      const responses = await fetchFile(args.REPL, urls, headersForTableRequest, true)
 
       // then dissect it into errors and non-errors
       const { errors, ok } = await handleErrors(responses, formatUrl, kind, args.REPL)
@@ -116,12 +122,12 @@ export default async function getDirect(args: Arguments<KubeOptions>, _kind: Pro
       }, undefined)
 
       if (args.execOptions.type === ExecType.TopLevel && metaTable.rows.length === 0 && !isWatchRequest(args)) {
-        return `No resources found in **${namespace}** namespace.`
+        return `No resources found in **${await getNamespace(args)}** namespace.`
       } else {
         try {
           // withNotFound will add error rows to the table for each error
           const table = withNotFound(await toKuiTable(metaTable, kind, args), errors.map(_ => _.message).join('\n'))
-          return table
+          return !isWatchRequest(args) ? table : makeWatchable(args, kind, table, formatUrl)
         } catch (err) {
           console.error('error formatting table', err)
           throw new Error('Internal Error')