feat: stream logs to Garden Cloud via gRPC (#7793)

* chore: change createActionLog signature

* feat: stream logs to Garden Cloud via gRPC

---------

Co-authored-by: Jon Edvald <edvald@gmail.com>

Author: edvald-garden

core/package.json

@@ -26,7 +26,7 @@
   "types": "build/src/index.d.ts",
   "dependencies": {
     "@bufbuild/protobuf": "2.8.0",
-    "@buf/garden_grow-platform.bufbuild_es": "2.8.0-20251027135652-778164cdd988.1",
+    "@buf/garden_grow-platform.bufbuild_es": "2.8.0-00000000000000-b98a6468c359.1",
     "@codenamize/codenamize": "^1.1.1",
     "@connectrpc/connect": "2.1.0",
     "@connectrpc/connect-node": "2.0.4",

core/src/actions/base.ts

@@ -757,8 +757,7 @@ export abstract class BaseAction<
 
     return createActionLog({
       log,
-      actionKind: this.kind,
-      actionName: this.name,
+      action: this,
     })
   }
 

core/src/actions/helpers.ts

@@ -110,7 +110,7 @@ export async function getActionState(
   graph: ResolvedConfigGraph,
   log: Log
 ): Promise<ActionState> {
-  const actionLog = createActionLog({ log, actionName: action.name, actionKind: action.kind })
+  const actionLog = createActionLog({ log, action })
   switch (action.kind) {
     case "Build":
       return (await router.build.getStatus({ action: action as ResolvedBuildAction, log: actionLog, graph }))?.result
@@ -156,7 +156,7 @@ export async function getDeployStatusPayloads({
     await Promise.all(
       actions.map(async (action) => {
         const startedAt = new Date().toISOString()
-        const actionLog = createActionLog({ log, actionName: action.name, actionKind: action.kind })
+        const actionLog = createActionLog({ log, action })
         const { result } = await router.deploy.getStatus({ action, log: actionLog, graph })
 
         const payload = makeActionCompletePayload({
@@ -194,7 +194,7 @@ export async function getBuildStatusPayloads({
     await Promise.all(
       actions.map(async (action) => {
         const startedAt = new Date().toISOString()
-        const actionLog = createActionLog({ log, actionName: action.name, actionKind: action.kind })
+        const actionLog = createActionLog({ log, action })
         const { result } = await router.build.getStatus({ action, log: actionLog, graph })
 
         const payload = makeActionCompletePayload({
@@ -230,7 +230,7 @@ export async function getTestStatusPayloads({
   return fromPairs(
     await Promise.all(
       actions.map(async (action) => {
-        const actionLog = createActionLog({ log, actionName: action.name, actionKind: action.kind })
+        const actionLog = createActionLog({ log, action })
         const startedAt = new Date().toISOString()
         const { result } = await router.test.getResult({ action, log: actionLog, graph })
         const payload = makeActionCompletePayload({
@@ -266,7 +266,7 @@ export async function getRunStatusPayloads({
   return fromPairs(
     await Promise.all(
       actions.map(async (action) => {
-        const actionLog = createActionLog({ log, actionName: action.name, actionKind: action.kind })
+        const actionLog = createActionLog({ log, action })
         const startedAt = new Date().toISOString()
         const { result } = await router.run.getResult({ action, log: actionLog, graph })
 

core/src/cli/cli.ts

@@ -312,6 +312,9 @@ ${renderCommands(commands)}
           }
         }
 
+        // Wire root logger Garden instance event bus
+        log.root.events = garden.events
+
         analytics = await garden.getAnalyticsHandler()
 
         // Register log file writers. We need to do this after the Garden class is initialised because

core/src/cloud/api/grpc-event-converter.ts

@@ -16,7 +16,8 @@ import {
 } from "@buf/garden_grow-platform.bufbuild_es/garden/public/events/v1/events_pb.js"
 import type { EventName as CoreEventName, EventPayload as CoreEventPayload } from "../../events/events.js"
 import type { GardenWithNewBackend } from "../../garden.js"
-import type { Log } from "../../logger/log-entry.js"
+import type { LogSymbol, Msg } from "../../logger/log-entry.js"
+import { isActionLogContext, isCoreLogContext, type Log } from "../../logger/log-entry.js"
 import { monotonicFactory, ulid, type ULID, type UUID } from "ulid"
 import { create } from "@bufbuild/protobuf"
 import {
@@ -49,6 +50,12 @@ import {
   DeployRunResultSchema,
 } from "@buf/garden_grow-platform.bufbuild_es/garden/public/events/v1/garden_action_pb.js"
 import type { DeployStatusForEventPayload } from "../../types/service.js"
+import {
+  DataFormat,
+  GardenLogMessageEmittedSchema,
+  LogSymbol as GrpcLogSymbol,
+} from "@buf/garden_grow-platform.bufbuild_es/garden/public/events/v1/garden_logs_pb.js"
+import type { LogEntryEventPayload } from "../api-legacy/restful-event-stream.js"
 
 const nextEventUlid = monotonicFactory()
 
@@ -73,6 +80,7 @@ const aecEnvironmentUpdateActionTriggeredMap = {
 export class GrpcEventConverter {
   private readonly garden: GardenWithNewBackend
   private readonly log: Log
+  private readonly shouldStreamLogEntries: boolean
 
   /**
    * It is important to keep it static,
@@ -81,9 +89,10 @@ export class GrpcEventConverter {
    */
   private static readonly uuidToUlidMap = new Map<UUID, ULID>()
 
-  constructor(garden: GardenWithNewBackend, log: Log) {
+  constructor(garden: GardenWithNewBackend, log: Log, shouldStreamLogEntries: boolean) {
     this.garden = garden
     this.log = log
+    this.shouldStreamLogEntries = shouldStreamLogEntries
   }
 
   convert<T extends CoreEventName>(name: T, payload: CoreEventPayload<T>): GrpcEventEnvelope[] {
@@ -132,18 +141,66 @@ export class GrpcEventConverter {
           payload: payload as CoreEventPayload<"aecAgentStatus">,
         })
         break
+      // NOTE: We're not propagating "log" events, only keeping those for legacy Cloud
+      case "logEntry":
+        events = this.handleLogEntry({ context, payload: payload as CoreEventPayload<"logEntry"> })
+        break
       default:
-        // TODO: handle all event cases
-        // name satisfies never // ensure all cases are handled
-        this.log.silly(`GrpcEventStream: Unhandled core event ${name}`)
         return []
     }
 
-    if (events.length === 0) {
-      this.log.silly(`GrpcEventStream: Ignoring core event ${name}`)
+    return events
+  }
+
+  private handleLogEntry({
+    context,
+    payload,
+  }: {
+    context: GardenEventContext
+    payload: LogEntryEventPayload
+  }): GrpcEventEnvelope[] {
+    if (!this.shouldStreamLogEntries) {
+      return []
     }
+    const msg = resolveMsg(payload.message.msg)
+    let rawMsg = resolveMsg(payload.message.rawMsg)
 
-    return events
+    if (msg === rawMsg) {
+      // No need to send both if they're the same
+      rawMsg = undefined
+    }
+
+    const coreLog = isCoreLogContext(payload.context) ? payload.context : undefined
+    const actionLog = isActionLogContext(payload.context) ? payload.context : undefined
+
+    let actionUlid: ULID | undefined = undefined
+    if (actionLog) {
+      actionUlid = this.mapToUlid(actionLog.actionUid, "actionUid", "actionUlid")
+    }
+
+    return [
+      createGardenCliEvent(context, GardenCliEventType.LOGS_EMITTED, {
+        case: "logsEmitted",
+        value: create(GardenLogMessageEmittedSchema, {
+          actionUlid,
+          loggedAt: timestampFromDate(new Date(payload.timestamp)),
+          logLevel: payload.level + 1,
+          // NOTE: We deprecated the sectionName and logMessage fields, preferring the logDetails field instead
+          originDescription: payload.context.origin,
+          logDetails: {
+            // Empty strings should be omitted
+            msg: msg ?? undefined,
+            rawMsg: rawMsg ?? undefined,
+            data: payload.message.data ?? undefined,
+            dataFormat: convertLogMessageDataFormat(payload.message.dataFormat),
+            symbol: convertLogSymbol(payload.message.symbol),
+            error: payload.message.error,
+            coreLog,
+            actionLog,
+          },
+        }),
+      }),
+    ]
   }
 
   private handleAecEnvironmentUpdate({
@@ -530,3 +587,38 @@ export function createGardenCliEvent(
 export function describeGrpcEvent(event: GrpcEventEnvelope): string {
   return `GrpcEvent(${event.eventUlid}, ${event.eventData.case}, ${event.eventData.value?.eventData.case})`
 }
+
+function convertLogSymbol(symbol: LogSymbol | undefined): GrpcLogSymbol | undefined {
+  switch (symbol) {
+    case "info":
+      return GrpcLogSymbol.INFO
+    case "success":
+      return GrpcLogSymbol.SUCCESS
+    case "warning":
+      return GrpcLogSymbol.WARNING
+    case "error":
+      return GrpcLogSymbol.ERROR
+    case "empty":
+      return GrpcLogSymbol.UNSPECIFIED
+    default:
+      return undefined
+  }
+}
+
+function convertLogMessageDataFormat(dataFormat: "json" | "yaml" | undefined): DataFormat | undefined {
+  switch (dataFormat) {
+    case "json":
+      return DataFormat.JSON
+    case "yaml":
+      return DataFormat.YAML
+    default:
+      return undefined
+  }
+}
+
+function resolveMsg(msg: Msg | undefined): string | undefined {
+  if (typeof msg === "function") {
+    return msg()
+  }
+  return msg
+}

core/src/cloud/api/grpc-event-stream.ts

@@ -11,8 +11,6 @@ import type { GardenWithNewBackend } from "../../garden.js"
 import { registerCleanupFunction, sleep } from "../../util/util.js"
 import type { Log } from "../../logger/log-entry.js"
 import type { EventName, EventPayload, GardenEventAnyListener } from "../../events/events.js"
-import { LogLevel } from "../../logger/logger.js"
-import type { LogEntryEventPayload } from "../api-legacy/restful-event-stream.js"
 import type {
   Event as GrpcEvent,
   GardenEventIngestionService,
@@ -71,7 +69,7 @@ export class GrpcEventStream {
     this.isClosed = false
     this.shouldStreamLogEntries = shouldStreamLogEntries
 
-    this.converter = new GrpcEventConverter(this.garden, this.log)
+    this.converter = new GrpcEventConverter(this.garden, this.log, this.shouldStreamLogEntries)
 
     // TODO: make sure it waits for the callback function completion
     registerCleanupFunction("grow-stream-session-cancelled-event", () => {
@@ -80,17 +78,19 @@ export class GrpcEventStream {
       }
 
       this.handleEvent("sessionCancelled", {})
-      this.close().catch(() => {})
+      this.close().catch(() => {
+        return
+      })
     })
 
-    this.logListener = (name, payload) => {
-      if (name === "logEntry" && payload.level <= LogLevel.debug) {
-        this.handleLogEntry(payload)
-      }
+    // Handle log entries in the converter
+    this.logListener = () => {
+      return
     }
+
     this.log.root.events.onAny(this.logListener)
 
-    this.eventListener = async (name, payload) => {
+    this.eventListener = (name, payload) => {
       this.handleEvent(name, payload)
     }
     this.garden.events.onAny(this.eventListener)
@@ -167,7 +167,15 @@ export class GrpcEventStream {
   }
 
   private handleEvent<T extends EventName>(name: T, payload: EventPayload<T>) {
-    const events = this.converter.convert(name, payload)
+    let events: GrpcEvent[] = []
+
+    try {
+      events = this.converter.convert(name, payload)
+    } catch (err) {
+      this.log.warn(`GrpcEventStream: Error while converting event ${name}: ${err}`)
+      return
+    }
+
     for (const event of events) {
       this.log.silly(
         () => `GrpcEventStream: ${this.outputStream ? "Sending" : "Buffering"} event ${describeGrpcEvent(event)}`
@@ -190,14 +198,6 @@ export class GrpcEventStream {
     }
   }
 
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  private handleLogEntry(logEntry: LogEntryEventPayload) {
-    if (!this.shouldStreamLogEntries) {
-      return
-    }
-    // TODO: logs handling
-  }
-
   private async streamEvents() {
     this.outputStream = createWritableIterable<IngestEventsRequest>()
 
@@ -235,7 +235,7 @@ export class GrpcEventStream {
             )}`
         )
       } else {
-        this.log.debug(() => `GrpcEventStream: Received ack for event ${nextAck.eventUlid}, final=${nextAck.final}`)
+        this.log.silly(() => `GrpcEventStream: Received ack for event ${nextAck.eventUlid}, final=${nextAck.final}`)
       }
 
       // Remove acknowledged event from the buffer
@@ -293,7 +293,7 @@ export class GrpcEventStream {
       return
     }
 
-    this.log.debug(() => `GrpcEventStream: Flushing ${this.eventBuffer.size} events from the buffer`)
+    this.log.silly(() => `GrpcEventStream: Flushing ${this.eventBuffer.size} events from the buffer`)
 
     // NOTE: The Map implementation in the javascript runtime guarantees that values will be iterated in the order they were added (FIFO).
     for (const event of this.eventBuffer.values()) {

core/src/commands/base.ts

@@ -180,7 +180,8 @@ export abstract class Command<
   noProject = false
   protected = false
   streamEvents = false // Set to true to stream events for the command
-  streamLogEntries = false // Set to true to stream log entries for the command
+  streamLogEntries = false // Set to true to stream log entries for the command to Garden Cloud v1 and v2
+  streamLogEntriesV2 = false // Set to true to stream log entries for the command to just Garden Cloud v2
   isCustom = false // Used to identify custom commands
   isDevCommand = false // Set to true for internal commands in interactive command-line commands
   ignoreOptions = false // Completely ignore all option flags and pass all arguments directly to the command
@@ -344,11 +345,14 @@ export abstract class Command<
 
         let result: CommandResult<R>
 
+        // We're streaming more logs to Garden Cloud v2 so we have a separate flag for that
+        const shouldStreamLogs = this.streamLogEntries || (!!garden.cloudApi && this.streamLogEntriesV2)
+
         const cloudEventStream = createCloudEventStream({
           sessionId: garden.sessionId,
           log,
           garden,
-          opts: { shouldStreamEvents: this.streamEvents, shouldStreamLogs: this.streamLogEntries },
+          opts: { shouldStreamEvents: this.streamEvents, shouldStreamLogs },
         })
         if (cloudEventStream) {
           log.silly(() => `Connecting Garden instance events to Cloud API`)

core/src/commands/build.ts

@@ -51,6 +51,7 @@ export class BuildCommand extends Command<Args, Opts> {
 
   override protected = true
   override streamEvents = true
+  override streamLogEntriesV2 = true
 
   override description = dedent`
     Runs all or specified Builds, taking into account build dependency order.

core/src/commands/custom.ts

@@ -73,7 +73,7 @@ export class CustomCommandWrapper extends Command {
   help = ""
 
   override isCustom = true
-
+  override streamLogEntriesV2 = true
   override allowUndefinedArguments = true
 
   constructor(public spec: CommandResource) {

core/src/commands/delete.ts

@@ -65,6 +65,7 @@ export class DeleteEnvironmentCommand extends Command<{}, DeleteEnvironmentOpts>
 
   override protected = true
   override streamEvents = true
+  override streamLogEntriesV2 = true
 
   override options = deleteEnvironmentOpts
 
@@ -149,6 +150,7 @@ export class DeleteDeployCommand extends Command<DeleteDeployArgs, DeleteDeployO
   override protected = true
   workflows = true
   override streamEvents = true
+  override streamLogEntriesV2 = true
 
   override options = deleteDeployOpts