Merge pull request #10637 from Budibase/merge-master

Merge master -> develop

lerna.json

@@ -1,5 +1,5 @@
 {
-  "version": "2.6.16-alpha.5",
+  "version": "2.6.18",
   "npmClient": "yarn",
   "packages": [
     "packages/backend-core",

packages/backend-core/src/context/mainContext.ts

@@ -104,6 +104,22 @@ async function newContext(updates: ContextMap, task: any) {
   return Context.run(context, task)
 }
 
+export async function doInAutomationContext(params: {
+  appId: string
+  automationId: string
+  task: any
+}): Promise<any> {
+  const tenantId = getTenantIDFromAppID(params.appId)
+  return newContext(
+    {
+      tenantId,
+      appId: params.appId,
+      automationId: params.automationId,
+    },
+    params.task
+  )
+}
+
 export async function doInContext(appId: string, task: any): Promise<any> {
   const tenantId = getTenantIDFromAppID(appId)
   return newContext(
@@ -187,6 +203,11 @@ export function getTenantId(): string {
   return tenantId
 }
 
+export function getAutomationId(): string | undefined {
+  const context = Context.get()
+  return context?.automationId
+}
+
 export function getAppId(): string | undefined {
   const context = Context.get()
   const foundId = context?.appId

packages/backend-core/src/context/types.ts

@@ -7,4 +7,5 @@ export type ContextMap = {
   identity?: IdentityContext
   environmentVariables?: Record<string, string>
   isScim?: boolean
+  automationId?: string
 }

packages/backend-core/src/logging/pino/logger.ts

@@ -39,6 +39,7 @@ if (!env.DISABLE_PINO_LOGGER) {
     objects?: any[]
     tenantId?: string
     appId?: string
+    automationId?: string
     identityId?: string
     identityType?: IdentityType
     correlationId?: string
@@ -86,18 +87,44 @@ if (!env.DISABLE_PINO_LOGGER) {
       contextObject = {
         tenantId: getTenantId(),
         appId: getAppId(),
+        automationId: getAutomationId(),
         identityId: identity?._id,
         identityType: identity?.type,
         correlationId: correlation.getId(),
       }
     }
 
-    const mergingObject = {
-      objects: objects.length ? objects : undefined,
+    const mergingObject: any = {
       err: error,
       ...contextObject,
     }
 
+    if (objects.length) {
+      // init generic data object for params supplied that don't have a
+      // '_logKey' field. This prints an object using argument index as the key
+      // e.g. { 0: {}, 1: {} }
+      const data: any = {}
+      let dataIndex = 0
+
+      for (let i = 0; i < objects.length; i++) {
+        const object = objects[i]
+        // the object has specified a log key
+        // use this instead of generic key
+        const logKey = object._logKey
+        if (logKey) {
+          delete object._logKey
+          mergingObject[logKey] = object
+        } else {
+          data[dataIndex] = object
+          dataIndex++
+        }
+      }
+
+      if (Object.keys(data).length) {
+        mergingObject.data = data
+      }
+    }
+
     return [mergingObject, message]
   }
 
@@ -159,6 +186,16 @@ if (!env.DISABLE_PINO_LOGGER) {
     return appId
   }
 
+  const getAutomationId = () => {
+    let appId
+    try {
+      appId = context.getAutomationId()
+    } catch (e) {
+      // do nothing
+    }
+    return appId
+  }
+
   const getIdentity = () => {
     let identity
     try {

packages/backend-core/src/queue/inMemoryQueue.ts

@@ -128,6 +128,7 @@ class InMemoryQueue {
 
   on() {
     // do nothing
+    return this
   }
 
   async waitForCompletion() {

packages/backend-core/src/queue/listeners.ts

@@ -1,5 +1,6 @@
 import { Job, JobId, Queue } from "bull"
 import { JobQueue } from "./constants"
+import * as context from "../context"
 
 export type StalledFn = (job: Job) => Promise<void>
 
@@ -31,77 +32,164 @@ function handleStalled(queue: Queue, removeStalledCb?: StalledFn) {
   })
 }
 
-function logging(queue: Queue, jobQueue: JobQueue) {
-  let eventType: string
-  switch (jobQueue) {
-    case JobQueue.AUTOMATION:
-      eventType = "automation-event"
-      break
-    case JobQueue.APP_BACKUP:
-      eventType = "app-backup-event"
-      break
-    case JobQueue.AUDIT_LOG:
-      eventType = "audit-log-event"
-      break
-    case JobQueue.SYSTEM_EVENT_QUEUE:
-      eventType = "system-event"
-      break
+function getLogParams(
+  eventType: QueueEventType,
+  event: BullEvent,
+  opts: {
+    job?: Job
+    jobId?: JobId
+    error?: Error
+  } = {},
+  extra: any = {}
+) {
+  const message = `[BULL] ${eventType}=${event}`
+  const err = opts.error
+
+  const bullLog = {
+    _logKey: "bull",
+    eventType,
+    event,
+    job: opts.job,
+    jobId: opts.jobId || opts.job?.id,
+    ...extra,
   }
+
+  let automationLog
+  if (opts.job?.data?.automation) {
+    automationLog = {
+      _logKey: "automation",
+      trigger: opts.job
+        ? opts.job.data.automation.definition.trigger.event
+        : undefined,
+    }
+  }
+
+  return [message, err, bullLog, automationLog]
+}
+
+enum BullEvent {
+  ERROR = "error",
+  WAITING = "waiting",
+  ACTIVE = "active",
+  STALLED = "stalled",
+  PROGRESS = "progress",
+  COMPLETED = "completed",
+  FAILED = "failed",
+  PAUSED = "paused",
+  RESUMED = "resumed",
+  CLEANED = "cleaned",
+  DRAINED = "drained",
+  REMOVED = "removed",
+}
+
+enum QueueEventType {
+  AUTOMATION_EVENT = "automation-event",
+  APP_BACKUP_EVENT = "app-backup-event",
+  AUDIT_LOG_EVENT = "audit-log-event",
+  SYSTEM_EVENT = "system-event",
+}
+
+const EventTypeMap: { [key in JobQueue]: QueueEventType } = {
+  [JobQueue.AUTOMATION]: QueueEventType.AUTOMATION_EVENT,
+  [JobQueue.APP_BACKUP]: QueueEventType.APP_BACKUP_EVENT,
+  [JobQueue.AUDIT_LOG]: QueueEventType.AUDIT_LOG_EVENT,
+  [JobQueue.SYSTEM_EVENT_QUEUE]: QueueEventType.SYSTEM_EVENT,
+}
+
+function logging(queue: Queue, jobQueue: JobQueue) {
+  const eventType = EventTypeMap[jobQueue]
+
+  function doInJobContext(job: Job, task: any) {
+    // if this is an automation job try to get the app id
+    const appId = job.data.event?.appId
+    if (appId) {
+      return context.doInContext(appId, task)
+    } else {
+      task()
+    }
+  }
+
+  queue
+    .on(BullEvent.STALLED, async (job: Job) => {
+      // A job has been marked as stalled. This is useful for debugging job
+      // workers that crash or pause the event loop.
+      await doInJobContext(job, () => {
+        console.error(...getLogParams(eventType, BullEvent.STALLED, { job }))
+      })
+    })
+    .on(BullEvent.ERROR, (error: any) => {
+      // An error occurred.
+      console.error(...getLogParams(eventType, BullEvent.ERROR, { error }))
+    })
+
   if (process.env.NODE_DEBUG?.includes("bull")) {
     queue
-      .on("error", (error: any) => {
-        // An error occurred.
-        console.error(`${eventType}=error error=${JSON.stringify(error)}`)
-      })
-      .on("waiting", (jobId: JobId) => {
+      .on(BullEvent.WAITING, (jobId: JobId) => {
         // A Job is waiting to be processed as soon as a worker is idling.
-        console.log(`${eventType}=waiting jobId=${jobId}`)
+        console.info(...getLogParams(eventType, BullEvent.WAITING, { jobId }))
       })
-      .on("active", (job: Job, jobPromise: any) => {
+      .on(BullEvent.ACTIVE, async (job: Job, jobPromise: any) => {
         // A job has started. You can use `jobPromise.cancel()`` to abort it.
-        console.log(`${eventType}=active jobId=${job.id}`)
+        await doInJobContext(job, () => {
+          console.info(...getLogParams(eventType, BullEvent.ACTIVE, { job }))
+        })
       })
-      .on("stalled", (job: Job) => {
-        // A job has been marked as stalled. This is useful for debugging job
-        // workers that crash or pause the event loop.
-        console.error(
-          `${eventType}=stalled jobId=${job.id} job=${JSON.stringify(job)}`
-        )
+      .on(BullEvent.PROGRESS, async (job: Job, progress: any) => {
+        // A job's progress was updated
+        await doInJobContext(job, () => {
+          console.info(
+            ...getLogParams(
+              eventType,
+              BullEvent.PROGRESS,
+              { job },
+              { progress }
+            )
+          )
+        })
       })
-      .on("progress", (job: Job, progress: any) => {
-        // A job's progress was updated!
-        console.log(
-          `${eventType}=progress jobId=${job.id} progress=${progress}`
-        )
-      })
-      .on("completed", (job: Job, result) => {
+      .on(BullEvent.COMPLETED, async (job: Job, result) => {
         // A job successfully completed with a `result`.
-        console.log(`${eventType}=completed jobId=${job.id} result=${result}`)
+        await doInJobContext(job, () => {
+          console.info(
+            ...getLogParams(eventType, BullEvent.COMPLETED, { job }, { result })
+          )
+        })
       })
-      .on("failed", (job, err: any) => {
+      .on(BullEvent.FAILED, async (job: Job, error: any) => {
         // A job failed with reason `err`!
-        console.log(`${eventType}=failed jobId=${job.id} error=${err}`)
+        await doInJobContext(job, () => {
+          console.error(
+            ...getLogParams(eventType, BullEvent.FAILED, { job, error })
+          )
+        })
       })
-      .on("paused", () => {
+      .on(BullEvent.PAUSED, () => {
         // The queue has been paused.
-        console.log(`${eventType}=paused`)
+        console.info(...getLogParams(eventType, BullEvent.PAUSED))
       })
-      .on("resumed", (job: Job) => {
+      .on(BullEvent.RESUMED, () => {
         // The queue has been resumed.
-        console.log(`${eventType}=paused jobId=${job.id}`)
+        console.info(...getLogParams(eventType, BullEvent.RESUMED))
       })
-      .on("cleaned", (jobs: Job[], type: string) => {
+      .on(BullEvent.CLEANED, (jobs: Job[], type: string) => {
         // Old jobs have been cleaned from the queue. `jobs` is an array of cleaned
         // jobs, and `type` is the type of jobs cleaned.
-        console.log(`${eventType}=cleaned length=${jobs.length} type=${type}`)
+        console.info(
+          ...getLogParams(
+            eventType,
+            BullEvent.CLEANED,
+            {},
+            { length: jobs.length, type }
+          )
+        )
       })
-      .on("drained", () => {
+      .on(BullEvent.DRAINED, () => {
         // Emitted every time the queue has processed all the waiting jobs (even if there can be some delayed jobs not yet processed)
-        console.log(`${eventType}=drained`)
+        console.info(...getLogParams(eventType, BullEvent.DRAINED))
       })
-      .on("removed", (job: Job) => {
+      .on(BullEvent.REMOVED, (job: Job) => {
         // A job successfully removed.
-        console.log(`${eventType}=removed jobId=${job.id}`)
+        console.info(...getLogParams(eventType, BullEvent.REMOVED, { job }))
       })
   }
 }

packages/server/src/automations/triggers.ts

@@ -9,7 +9,7 @@ import { checkTestFlag } from "../utilities/redis"
 import * as utils from "./utils"
 import env from "../environment"
 import { context, db as dbCore } from "@budibase/backend-core"
-import { Automation, Row } from "@budibase/types"
+import { Automation, Row, AutomationData, AutomationJob } from "@budibase/types"
 
 export const TRIGGER_DEFINITIONS = definitions
 const JOB_OPTS = {
@@ -109,14 +109,16 @@ export async function externalTrigger(
     }
     params.fields = coercedFields
   }
-  const data: Record<string, any> = { automation, event: params }
+
+  const data: AutomationData = { automation, event: params as any }
   if (getResponses) {
     data.event = {
       ...data.event,
       appId: context.getAppId(),
       automation,
     }
-    return utils.processEvent({ data })
+    const job = { data } as AutomationJob
+    return utils.processEvent(job)
   } else {
     return automationQueue.add(data, JOB_OPTS)
   }

packages/server/src/automations/utils.ts

@@ -8,7 +8,7 @@ import { db as dbCore, context } from "@budibase/backend-core"
 import { getAutomationMetadataParams } from "../db/utils"
 import { cloneDeep } from "lodash/fp"
 import { quotas } from "@budibase/pro"
-import { Automation, WebhookActionType } from "@budibase/types"
+import { Automation, AutomationJob, WebhookActionType } from "@budibase/types"
 import sdk from "../sdk"
 
 const REBOOT_CRON = "@reboot"
@@ -16,27 +16,40 @@ const WH_STEP_ID = definitions.WEBHOOK.stepId
 const CRON_STEP_ID = definitions.CRON.stepId
 const Runner = new Thread(ThreadType.AUTOMATION)
 
-const jobMessage = (job: any, message: string) => {
-  return `app=${job.data.event.appId} automation=${job.data.automation._id} jobId=${job.id} trigger=${job.data.automation.definition.trigger.event} : ${message}`
+function loggingArgs(job: AutomationJob) {
+  return [
+    {
+      _logKey: "automation",
+      trigger: job.data.automation.definition.trigger.event,
+    },
+    {
+      _logKey: "bull",
+      jobId: job.id,
+    },
+  ]
 }
 
-export async function processEvent(job: any) {
-  try {
-    const automationId = job.data.automation._id
-    console.log(jobMessage(job, "running"))
-    // need to actually await these so that an error can be captured properly
-    return await context.doInContext(job.data.event.appId, async () => {
+export async function processEvent(job: AutomationJob) {
+  const appId = job.data.event.appId!
+  const automationId = job.data.automation._id!
+  const task = async () => {
+    try {
+      // need to actually await these so that an error can be captured properly
+      console.log("automation running", ...loggingArgs(job))
+
       const runFn = () => Runner.run(job)
-      return quotas.addAutomation(runFn, {
+      const result = await quotas.addAutomation(runFn, {
         automationId,
       })
-    })
-  } catch (err) {
-    const errJson = JSON.stringify(err)
-    console.error(jobMessage(job, `was unable to run - ${errJson}`))
-    console.trace(err)
-    return { err }
+      console.log("automation completed", ...loggingArgs(job))
+      return result
+    } catch (err) {
+      console.error(`automation was unable to run`, err, ...loggingArgs(job))
+      return { err }
+    }
   }
+
+  return await context.doInAutomationContext({ appId, automationId, task })
 }
 
 export async function updateTestHistory(

packages/server/src/definitions/automations.ts

@@ -1,4 +1,4 @@
-import { AutomationResults, AutomationStep, Document } from "@budibase/types"
+import { AutomationResults, AutomationStep } from "@budibase/types"
 
 export enum LoopStepType {
   ARRAY = "Array",
@@ -27,7 +27,3 @@ export interface AutomationContext extends AutomationResults {
   env?: Record<string, string>
   trigger: any
 }
-
-export interface AutomationMetadata extends Document {
-  errorCount?: number
-}

packages/server/src/threads/automation.ts

@@ -13,13 +13,18 @@ import { generateAutomationMetadataID, isProdAppID } from "../db/utils"
 import { definitions as triggerDefs } from "../automations/triggerInfo"
 import { AutomationErrors, MAX_AUTOMATION_RECURRING_ERRORS } from "../constants"
 import { storeLog } from "../automations/logging"
-import { Automation, AutomationStep, AutomationStatus } from "@budibase/types"
+import {
+  Automation,
+  AutomationStep,
+  AutomationStatus,
+  AutomationMetadata,
+  AutomationJob,
+} from "@budibase/types"
 import {
   LoopStep,
   LoopInput,
   TriggerOutput,
   AutomationContext,
-  AutomationMetadata,
 } from "../definitions/automations"
 import { WorkerCallback } from "./definitions"
 import { context, logging } from "@budibase/backend-core"
@@ -60,11 +65,11 @@ class Orchestrator {
   _job: Job
   executionOutput: AutomationContext
 
-  constructor(job: Job) {
-    let automation = job.data.automation,
-      triggerOutput = job.data.event
+  constructor(job: AutomationJob) {
+    let automation = job.data.automation
+    let triggerOutput = job.data.event
     const metadata = triggerOutput.metadata
-    this._chainCount = metadata ? metadata.automationChainCount : 0
+    this._chainCount = metadata ? metadata.automationChainCount! : 0
     this._appId = triggerOutput.appId as string
     this._job = job
     const triggerStepId = automation.definition.trigger.stepId

packages/server/src/threads/definitions.ts

@@ -1,5 +1,3 @@
-import { EnvironmentVariablesDecrypted } from "@budibase/types"
-
 export type WorkerCallback = (error: any, response?: any) => void
 
 export interface QueryEvent {

packages/server/src/threads/index.ts

@@ -1,5 +1,7 @@
 import workerFarm from "worker-farm"
 import env from "../environment"
+import { AutomationJob } from "@budibase/types"
+import { QueryEvent } from "./definitions"
 
 export const ThreadType = {
   QUERY: "query",
@@ -64,11 +66,11 @@ export class Thread {
     )
   }
 
-  run(data: any) {
+  run(job: AutomationJob | QueryEvent) {
     const timeout = this.timeoutMs
     return new Promise((resolve, reject) => {
       function fire(worker: any) {
-        worker.execute(data, (err: any, response: any) => {
+        worker.execute(job, (err: any, response: any) => {
           if (err && err.type === "TimeoutError") {
             reject(
               new Error(`Query response time exceeded ${timeout}ms timeout.`)

packages/types/src/documents/app/automation.ts

@@ -178,3 +178,8 @@ export type AutomationStepInput = {
   appId: string
   apiKey?: string
 }
+
+export interface AutomationMetadata extends Document {
+  errorCount?: number
+  automationChainCount?: number
+}

packages/types/src/sdk/automations/index.ts

@@ -0,0 +1,15 @@
+import { Automation, AutomationMetadata } from "../../documents"
+import { Job } from "bull"
+
+export interface AutomationDataEvent {
+  appId?: string
+  metadata?: AutomationMetadata
+  automation?: Automation
+}
+
+export interface AutomationData {
+  event: AutomationDataEvent
+  automation: Automation
+}
+
+export type AutomationJob = Job<AutomationData>

packages/types/src/sdk/index.ts

@@ -1,3 +1,4 @@
+export * from "./automations"
 export * from "./hosting"
 export * from "./context"
 export * from "./events"