feat(internal): support persistent commands in WS API

core/src/commands/base.ts

@@ -66,6 +66,8 @@ interface PrepareOutput {
   persistent: boolean
 }
 
+type DataCallback = (data: any) => void
+
 export abstract class Command<T extends Parameters = {}, U extends Parameters = {}> {
   abstract name: string
   abstract help: string
@@ -87,7 +89,13 @@ export abstract class Command<T extends Parameters = {}, U extends Parameters =
   streamLogEntries: boolean = false // Set to true to stream log entries for the command
   server: GardenServer | undefined = undefined
 
+  subscribers: DataCallback[]
+  terminated: boolean
+
   constructor(private parent?: CommandGroup) {
+    this.subscribers = []
+    this.terminated = false
+
     // Make sure arguments and options don't have overlapping key names.
     if (this.arguments && this.options) {
       for (const key of Object.keys(this.options)) {
@@ -168,6 +176,34 @@ export abstract class Command<T extends Parameters = {}, U extends Parameters =
     return { persistent: false }
   }
 
+  /**
+   * Called by e.g. the WebSocket server to terminate persistent commands.
+   */
+  terminate() {
+    this.terminated = true
+  }
+
+  /**
+   * Subscribe to any data emitted by commands via the .emit() method
+   */
+  subscribe(cb: (data: string) => void) {
+    this.subscribers.push(cb)
+  }
+
+  /**
+   * Emit data to all subscribers
+   */
+  emit(log: LogEntry, data: string) {
+    for (const subscriber of this.subscribers) {
+      // Ignore any errors here
+      try {
+        subscriber(data)
+      } catch (err) {
+        log.debug(`Error when calling subscriber on ${this.getFullName()} command: ${err.message}`)
+      }
+    }
+  }
+
   abstract printHeader(params: PrintHeaderParams<T, U>): void
 
   // Note: Due to a current TS limitation (apparently covered by https://github.com/Microsoft/TypeScript/issues/7011),

core/src/commands/build.ts

@@ -23,6 +23,7 @@ import { startServer } from "../server/server"
 import { flatten } from "lodash"
 import { BuildTask } from "../tasks/build"
 import { StringsParameter, BooleanParameter } from "../cli/params"
+import { Garden } from "../garden"
 
 const buildArgs = {
   modules: new StringsParameter({
@@ -65,12 +66,12 @@ export class BuildCommand extends Command<Args, Opts> {
   arguments = buildArgs
   options = buildOpts
 
-  outputsSchema = () => processCommandResultSchema()
+  private garden?: Garden
 
-  private isPersistent = (opts) => !!opts.watch
+  outputsSchema = () => processCommandResultSchema()
 
   async prepare({ footerLog, opts }: PrepareParams<Args, Opts>) {
-    const persistent = this.isPersistent(opts)
+    const persistent = !!opts.watch
 
     if (persistent) {
       this.server = await startServer({ log: footerLog })
@@ -79,6 +80,10 @@ export class BuildCommand extends Command<Args, Opts> {
     return { persistent }
   }
 
+  terminate() {
+    this.garden?.events.emit("_exit", {})
+  }
+
   printHeader({ headerLog }) {
     printHeader(headerLog, "Build", "hammer")
   }
@@ -90,6 +95,8 @@ export class BuildCommand extends Command<Args, Opts> {
     args,
     opts,
   }: CommandParams<Args, Opts>): Promise<CommandResult<ProcessCommandResult>> {
+    this.garden = garden
+
     if (this.server) {
       this.server.setGarden(garden)
     }

core/src/commands/dashboard.ts

@@ -82,8 +82,10 @@ export class DashboardCommand extends Command<Args, Opts> {
     this.server!.setGarden(garden)
 
     // The server doesn't block, so we need to loop indefinitely here.
-    while (true) {
-      await sleep(10000)
+    while (!this.terminated) {
+      await sleep(1000)
     }
+
+    return {}
   }
 }

core/src/commands/deploy.ts

@@ -28,6 +28,7 @@ import { DeployTask } from "../tasks/deploy"
 import { naturalList } from "../util/string"
 import chalk = require("chalk")
 import { StringsParameter, BooleanParameter, ParameterValues } from "../cli/params"
+import { Garden } from "../garden"
 
 export const deployArgs = {
   services: new StringsParameter({
@@ -100,6 +101,8 @@ export class DeployCommand extends Command<Args, Opts> {
   arguments = deployArgs
   options = deployOpts
 
+  private garden?: Garden
+
   outputsSchema = () => processCommandResultSchema()
 
   private isPersistent = (opts: ParameterValues<Opts>) => !!opts.watch || !!opts["hot-reload"] || !!opts["dev-mode"]
@@ -118,13 +121,19 @@ export class DeployCommand extends Command<Args, Opts> {
     return { persistent }
   }
 
+  terminate() {
+    this.garden?.events.emit("_exit", {})
+  }
+
   async action({
     garden,
     log,
     footerLog,
     args,
     opts,
   }: CommandParams<Args, Opts>): Promise<CommandResult<ProcessCommandResult>> {
+    this.garden = garden
+
     if (this.server) {
       this.server.setGarden(garden)
     }

core/src/commands/dev.ts

@@ -92,6 +92,8 @@ export class DevCommand extends Command<DevCommandArgs, DevCommandOpts> {
   arguments = devArgs
   options = devOpts
 
+  private garden?: Garden
+
   printHeader({ headerLog }) {
     printHeader(headerLog, "Dev", "keyboard")
   }
@@ -111,13 +113,18 @@ export class DevCommand extends Command<DevCommandArgs, DevCommandOpts> {
     return { persistent: true }
   }
 
+  terminate() {
+    this.garden?.events.emit("_exit", {})
+  }
+
   async action({
     garden,
     log,
     footerLog,
     args,
     opts,
   }: CommandParams<DevCommandArgs, DevCommandOpts>): Promise<CommandResult> {
+    this.garden = garden
     this.server?.setGarden(garden)
 
     const graph = await garden.getConfigGraph(log)

core/src/commands/logs.ts

@@ -6,7 +6,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  */
 
-import { Command, CommandResult, CommandParams } from "./base"
+import { Command, CommandResult, CommandParams, PrepareParams } from "./base"
 import chalk from "chalk"
 import { sortBy } from "lodash"
 import { ServiceLogEntry } from "../types/plugin/service/getServiceLogs"
@@ -20,6 +20,7 @@ import stripAnsi = require("strip-ansi")
 import hasAnsi = require("has-ansi")
 import { dedent } from "../util/string"
 import { padSection } from "../logger/renderers"
+import { PluginEventBroker } from "../plugin-context"
 
 const logsArgs = {
   services: new StringsParameter({
@@ -33,7 +34,6 @@ const logsOpts = {
   "follow": new BooleanParameter({
     help: "Continuously stream new logs from the service(s).",
     alias: "f",
-    cliOnly: true,
   }),
   "tail": new IntegerParameter({
     help: dedent`
@@ -99,6 +99,8 @@ export class LogsCommand extends Command<Args, Opts> {
   arguments = logsArgs
   options = logsOpts
 
+  private events?: PluginEventBroker
+
   getLoggerType(): LoggerType {
     return "basic"
   }
@@ -107,6 +109,14 @@ export class LogsCommand extends Command<Args, Opts> {
     printHeader(headerLog, "Logs", "scroll")
   }
 
+  async prepare({ opts }: PrepareParams<Args, Opts>) {
+    return { persistent: !!opts.follow }
+  }
+
+  terminate() {
+    this.events?.emit("abort", {})
+  }
+
   async action({ garden, log, args, opts }: CommandParams<Args, Opts>): Promise<CommandResult<ServiceLogEntry[]>> {
     const { follow, timestamps } = opts
     let tail = opts.tail as number | undefined
@@ -206,24 +216,26 @@ export class LogsCommand extends Command<Args, Opts> {
     }
 
     void stream.forEach((entry) => {
-      // Skip emtpy entries
+      // Skip empty entries
       if (skipEntry(entry)) {
         return
       }
 
       if (follow) {
         const levelStr = logLevelMap[entry.level || LogLevel.info] || "info"
         const msg = formatEntry(entry)
+        this.emit(log, JSON.stringify({ msg, timestamp: entry.timestamp?.getTime(), level: levelStr }))
         log[levelStr]({ msg })
       } else {
         result.push(entry)
       }
     })
 
     const actions = await garden.getActionRouter()
+    this.events = new PluginEventBroker()
 
     await Bluebird.map(services, async (service: GardenService<any>) => {
-      await actions.getServiceLogs({ log, graph, service, stream, follow, tail, since })
+      await actions.getServiceLogs({ log, graph, service, stream, follow, tail, since, events: this.events })
     })
 
     const sorted = sortBy(result, "timestamp")

core/src/commands/test.ts

@@ -26,6 +26,7 @@ import { printHeader } from "../logger/util"
 import { startServer } from "../server/server"
 import { StringsParameter, BooleanParameter } from "../cli/params"
 import { deline } from "../util/string"
+import { Garden } from "../garden"
 
 export const testArgs = {
   modules: new StringsParameter({
@@ -96,14 +97,14 @@ export class TestCommand extends Command<Args, Opts> {
 
   outputsSchema = () => processCommandResultSchema()
 
-  private isPersistent = (opts) => !!opts.watch
+  private garden?: Garden
 
   printHeader({ headerLog }) {
     printHeader(headerLog, `Running tests`, "thermometer")
   }
 
   async prepare({ footerLog, opts }: PrepareParams<Args, Opts>) {
-    const persistent = this.isPersistent(opts)
+    const persistent = !!opts.watch
 
     if (persistent) {
       this.server = await startServer({ log: footerLog })
@@ -112,13 +113,19 @@ export class TestCommand extends Command<Args, Opts> {
     return { persistent }
   }
 
+  terminate() {
+    this.garden?.events.emit("_exit", {})
+  }
+
   async action({
     garden,
     log,
     footerLog,
     args,
     opts,
   }: CommandParams<Args, Opts>): Promise<CommandResult<ProcessCommandResult>> {
+    this.garden = garden
+
     if (this.server) {
       this.server.setGarden(garden)
     }

core/src/plugin-context.ts

@@ -91,6 +91,7 @@ export const pluginContextSchema = () =>
     })
 
 interface PluginEvents {
+  abort: { reason?: string }
   log: { data: Buffer }
 }
 

core/src/plugins/kubernetes/logs.ts

@@ -52,6 +52,11 @@ export async function streamK8sLogs(params: GetAllLogsParams) {
 
   if (params.follow) {
     const logsFollower = new K8sLogFollower({ ...params, k8sApi: api })
+
+    params.ctx.events.on("abort", () => {
+      logsFollower.close()
+    })
+
     await logsFollower.followLogs({ tail: params.tail, since: params.since })
   } else {
     const pods = await getAllPods(api, params.defaultNamespace, params.resources)

core/src/server/commands.ts

@@ -12,7 +12,6 @@ import { Command, CommandGroup } from "../commands/base"
 import { joi } from "../config/common"
 import { validateSchema } from "../config/validation"
 import { extend, mapValues, omitBy } from "lodash"
-import { Garden } from "../garden"
 import { LogLevel } from "../logger/logger"
 import { LogEntry } from "../logger/log-entry"
 import { Parameters, ParameterValues, globalOptions } from "../cli/params"
@@ -37,20 +36,14 @@ const baseRequestSchema = () =>
   })
 
 /**
- * Validate and map a request body to a Command, execute its action, and return its result.
+ * Validate and map a request body to a Command
  */
-export async function resolveRequest(
-  ctx: Koa.ParameterizedContext,
-  garden: Garden,
-  log: LogEntry,
-  commands: CommandMap,
-  request: any
-) {
+export function parseRequest(ctx: Koa.ParameterizedContext, log: LogEntry, commands: CommandMap, request: any) {
   // Perform basic validation and find command.
   try {
     request = validateSchema(request, baseRequestSchema(), { context: "API request" })
-  } catch {
-    ctx.throw(400, "Invalid request format")
+  } catch (err) {
+    ctx.throw(400, "Invalid request format: " + err.message)
   }
 
   const commandSpec = commands[request.command]
@@ -76,18 +69,15 @@ export async function resolveRequest(
   const optParams = extend({ ...globalOptions, ...command.options })
   const cmdOpts = mapParams(ctx, request.parameters, optParams)
 
-  // TODO: validate result schema
-  return command.action({
-    garden,
+  return {
+    command,
     log: cmdLog,
-    headerLog: cmdLog,
-    footerLog: cmdLog,
     args: cmdArgs,
     opts: cmdOpts,
-  })
+  }
 }
 
-export async function prepareCommands(): Promise<CommandMap> {
+export function prepareCommands(): CommandMap {
   const commands: CommandMap = {}
 
   function addCommand(command: Command) {