fix(cloud): add env and ns IDs to event payloads

This enables a more optimized event processing pipeline in Garden Cloud.

core/src/cli/cli.ts

@@ -33,14 +33,13 @@ import { ERROR_LOG_FILENAME, DEFAULT_API_VERSION, DEFAULT_GARDEN_DIR_NAME, LOGS_
 import { generateBasicDebugInfoReport } from "../commands/get/get-debug-info"
 import { AnalyticsHandler } from "../analytics/analytics"
 import { defaultDotIgnoreFiles } from "../util/fs"
-import { BufferedEventStream } from "../cloud/buffered-event-stream"
+import { BufferedEventStream, ConnectBufferedEventStreamParams } from "../cloud/buffered-event-stream"
 import { GardenProcess } from "../db/entities/garden-process"
 import { DashboardEventStream } from "../server/dashboard-event-stream"
 import { GardenPluginCallback } from "../types/plugin/plugin"
 import { renderError } from "../logger/renderers"
 import { CloudApi } from "../cloud/api"
 import chalk = require("chalk")
-import { registerSession } from "../cloud/session-lifecycle"
 
 export async function makeDummyGarden(root: string, gardenOpts: GardenOpts) {
   const environments = gardenOpts.environmentName
@@ -282,21 +281,20 @@ ${renderCommands(commands)}
           dashboardEventStream.connect({ garden, ignoreHost: commandServerUrl, streamEvents, streamLogEntries })
           const runningServers = await dashboardEventStream.updateTargets()
 
+          if (cloudApi && !cloudApi.sessionRegistered && command.streamEvents) {
+            // Note: If a config change during a watch-mode command's execution results in the resolved environment
+            // and/or namespace name changing, we don't change the session ID, environment ID or namespace ID used when
+            // streaming events.
+            await cloudApi.registerSession({
+              sessionId,
+              commandInfo,
+              localServerPort: command.server?.port,
+              environment: garden.environmentName,
+              namespace: garden.namespace,
+            })
+          }
+
           if (persistent && command.server) {
-            if (cloudApi) {
-              // TODO: provide the environment & namespace IDs returned by this helper to `this.bufferedEventStream`,
-              // and include them in all log/event batches once the API is ready for / expects that.
-              await registerSession({
-                log,
-                cloudApi,
-                sessionId,
-                commandInfo,
-                localServerPort: command.server.port,
-                projectId: cloudApi.projectId,
-                environment: garden.environmentName,
-                namespace: garden.namespace,
-              })
-            }
             // If there is an explicit `garden dashboard` process running for the current project+env, and a server
             // is started in this Command, we show the URL to the external dashboard. Otherwise the built-in one.
             const dashboardProcess = GardenProcess.getDashboardProcess(runningServers, {
@@ -312,7 +310,7 @@ ${renderCommands(commands)}
 
         if (cloudApi) {
           log.silly(`Connecting Garden instance to GE BufferedEventStream`)
-          this.bufferedEventStream.connect({
+          const connectParams: ConnectBufferedEventStreamParams = {
             garden,
             streamEvents,
             streamLogEntries,
@@ -321,7 +319,8 @@ ${renderCommands(commands)}
                 enterprise: true,
               },
             ],
-          })
+          }
+          this.bufferedEventStream.connect(connectParams)
           if (streamEvents) {
             this.bufferedEventStream.streamEvent("commandInfo", commandInfo)
           }

core/src/cloud/api.ts

@@ -21,6 +21,7 @@ import { deline } from "../util/string"
 import chalk from "chalk"
 import { GetProjectResponse } from "@garden-io/platform-api-types"
 import { getCloudDistributionName, getPackageVersion } from "../util/util"
+import { CommandInfo } from "../plugin-context"
 
 const gardenClientName = "garden-core"
 const gardenClientVersion = getPackageVersion()
@@ -75,6 +76,12 @@ export type ApiFetchResponse<T> = T & {
   headers: IncomingHttpHeaders
 }
 
+// TODO: Read this from the `api-types` package once the session registration logic has been released in Cloud.
+export interface RegisterSessionResponse {
+  environmentId: number
+  namespaceId: number
+}
+
 /**
  * A helper function that finds a project without resolving template strings and returns the enterprise
  * config. Needed since the EnterpriseApi is generally used before initializing the Garden class.
@@ -110,6 +117,11 @@ export class CloudApi {
   public domain: string
   public projectId: string
 
+  // Set when/if the Core session is registered with Cloud
+  public environmentId?: number
+  public namespaceId?: number
+  public sessionRegistered = false
+
   constructor(log: LogEntry, enterpriseDomain: string, projectId: string) {
     this.log = log
     this.domain = enterpriseDomain
@@ -456,6 +468,54 @@ export class CloudApi {
     })
   }
 
+  async registerSession({
+    sessionId,
+    commandInfo,
+    localServerPort,
+    environment,
+    namespace,
+  }: {
+    sessionId: string
+    commandInfo: CommandInfo
+    localServerPort?: number
+    environment: string
+    namespace: string
+  }): Promise<void> {
+    try {
+      const body = {
+        sessionId,
+        commandInfo,
+        localServerPort,
+        projectUid: this.projectId,
+        environment,
+        namespace,
+      }
+      this.log.debug("Registering session with Garden Cloud.")
+      const res: RegisterSessionResponse = await this.post("sessions", {
+        body,
+        retry: true,
+        retryDescription: "Registering session",
+      })
+      this.environmentId = res.environmentId
+      this.namespaceId = res.namespaceId
+      this.log.debug("Successfully registered session with Garden Cloud.")
+    } catch (err) {
+      // We don't want the command to fail when an error occurs during session registration.
+      if (isGotError(err, 422)) {
+        const errMsg = deline`
+          Session registration skipped due to mismatch between CLI and API versions. Please make sure your Garden CLI
+          version is compatible with your version of Garden Cloud.
+        `
+        this.log.debug(errMsg)
+      } else {
+        // TODO: Reintroduce error-level warning when we're checking if the Cloud/Enterprise version is compatible with
+        // the Core version.
+        this.log.verbose(`An error occurred while registering the session: ${err.message}`)
+      }
+    }
+    this.sessionRegistered = true
+  }
+
   async getProject() {
     const res = await this.get<GetProjectResponse>(`/projects/uid/${this.projectId}`)
     return res.data

core/src/cloud/buffered-event-stream.ts

@@ -62,6 +62,8 @@ export interface ConnectBufferedEventStreamParams {
   targets?: StreamTarget[]
   streamEvents: boolean
   streamLogEntries: boolean
+  namespaceId?: number
+  environmentId?: number
   garden: Garden
 }
 
@@ -73,6 +75,10 @@ interface ApiBatchBase {
 
 export interface ApiEventBatch extends ApiBatchBase {
   events: StreamEvent[]
+  environmentId?: number
+  namespaceId?: number
+  // TODO: Remove the `environment` and `namespace` params once we no longer need to support Cloud/Enterprise
+  // versions that expect them.
   environment: string
   namespace: string
 }
@@ -234,6 +240,8 @@ export class BufferedEventStream {
       workflowRunUid: this.workflowRunUid,
       sessionId: this.sessionId,
       projectUid: this.garden.projectId || undefined,
+      environmentId: this.cloudApi?.environmentId,
+      namespaceId: this.cloudApi?.namespaceId,
       environment: this.garden.environmentName,
       namespace: this.garden.namespace,
     }