Merge branch 'master' into helm-fix

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

@@ -15,6 +15,7 @@ function newJob(queue: string, message: any) {
     timestamp: Date.now(),
     queue: queue,
     data: message,
+    opts: {},
   }
 }
 

packages/server/src/automations/utils.ts

@@ -16,6 +16,7 @@ import {
 } from "@budibase/types"
 import sdk from "../sdk"
 import { automationsEnabled } from "../features"
+import tracer from "dd-trace"
 
 const REBOOT_CRON = "@reboot"
 const WH_STEP_ID = definitions.WEBHOOK.stepId
@@ -39,8 +40,37 @@ function loggingArgs(job: AutomationJob) {
 }
 
 export async function processEvent(job: AutomationJob) {
+  return tracer.trace(
+    "processEvent",
+    { resource: "automation" },
+    async span => {
       const appId = job.data.event.appId!
       const automationId = job.data.automation._id!
+
+      span?.addTags({
+        appId,
+        automationId,
+        job: {
+          id: job.id,
+          name: job.name,
+          attemptsMade: job.attemptsMade,
+          opts: {
+            attempts: job.opts.attempts,
+            priority: job.opts.priority,
+            delay: job.opts.delay,
+            repeat: job.opts.repeat,
+            backoff: job.opts.backoff,
+            lifo: job.opts.lifo,
+            timeout: job.opts.timeout,
+            jobId: job.opts.jobId,
+            removeOnComplete: job.opts.removeOnComplete,
+            removeOnFail: job.opts.removeOnFail,
+            stackTraceLimit: job.opts.stackTraceLimit,
+            preventParsingData: job.opts.preventParsingData,
+          },
+        },
+      })
+
       const task = async () => {
         try {
           // need to actually await these so that an error can be captured properly
@@ -53,13 +83,20 @@ export async function processEvent(job: AutomationJob) {
           console.log("automation completed", ...loggingArgs(job))
           return result
         } catch (err) {
-      console.error(`automation was unable to run`, err, ...loggingArgs(job))
+          span?.addTags({ error: true })
+          console.error(
+            `automation was unable to run`,
+            err,
+            ...loggingArgs(job)
+          )
           return { err }
         }
       }
 
       return await context.doInAutomationContext({ appId, automationId, task })
     }
+  )
+}
 
 export async function updateTestHistory(
   appId: any,

packages/server/src/threads/automation.ts

@@ -34,6 +34,7 @@ import { cloneDeep } from "lodash/fp"
 import { performance } from "perf_hooks"
 import * as sdkUtils from "../sdk/utils"
 import env from "../environment"
+import tracer from "dd-trace"
 
 threadUtils.threadSetup()
 const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
@@ -242,6 +243,15 @@ class Orchestrator {
   }
 
   async execute(): Promise<any> {
+    return tracer.trace(
+      "Orchestrator.execute",
+      { resource: "automation" },
+      async span => {
+        span?.addTags({
+          appId: this._appId,
+          automationId: this._automation._id,
+        })
+
         // this will retrieve from context created at start of thread
         this._context.env = await sdkUtils.getEnvironmentVariables()
         let automation = this._automation
@@ -257,15 +267,39 @@ class Orchestrator {
         let timeout = this._job.data.event.timeout
         // check if this is a recurring automation,
         if (isProdAppID(this._appId) && isRecurring(automation)) {
+          span?.addTags({ recurring: true })
           metadata = await this.getMetadata()
           const shouldStop = await this.checkIfShouldStop(metadata)
           if (shouldStop) {
+            span?.addTags({ shouldStop: true })
             return
           }
         }
         const start = performance.now()
         for (let step of automation.definition.steps) {
+          const stepSpan = tracer.startSpan("Orchestrator.execute.step", {
+            childOf: span,
+          })
+          stepSpan.addTags({
+            resource: "automation",
+            step: {
+              stepId: step.stepId,
+              id: step.id,
+              name: step.name,
+              type: step.type,
+              title: step.stepTitle,
+              internal: step.internal,
+              deprecated: step.deprecated,
+            },
+          })
+
+          let input: any,
+            iterations = 1,
+            iterationCount = 0
+
+          try {
             if (timeoutFlag) {
+              span?.addTags({ timedOut: true })
               break
             }
 
@@ -276,10 +310,6 @@ class Orchestrator {
             }
 
             stepCount++
-      let input: any,
-        iterations = 1,
-        iterationCount = 0
-
             if (step.stepId === LOOP_STEP_ID) {
               loopStep = step
               loopStepNumber = stepCount
@@ -289,22 +319,31 @@ class Orchestrator {
             if (loopStep) {
               input = await processObject(loopStep.inputs, this._context)
               iterations = getLoopIterations(loopStep as LoopStep)
+              stepSpan?.addTags({ step: { iterations } })
             }
             for (let index = 0; index < iterations; index++) {
               let originalStepInput = cloneDeep(step.inputs)
               // Handle if the user has set a max iteration count or if it reaches the max limit set by us
               if (loopStep && input.binding) {
-          let tempOutput = { items: loopSteps, iterations: iterationCount }
+                let tempOutput = {
+                  items: loopSteps,
+                  iterations: iterationCount,
+                }
                 try {
                   loopStep.inputs.binding = automationUtils.typecastForLooping(
                     loopStep as LoopStep,
                     loopStep.inputs as LoopInput
                   )
                 } catch (err) {
-            this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
+                  this.updateContextAndOutput(
+                    loopStepNumber,
+                    step,
+                    tempOutput,
+                    {
                       status: AutomationErrors.INCORRECT_TYPE,
                       success: false,
-            })
+                    }
+                  )
                   loopSteps = undefined
                   loopStep = undefined
                   break
@@ -349,7 +388,8 @@ class Orchestrator {
                     }
                   } else {
                     if (typeof value === "string") {
-                originalStepInput[key] = automationUtils.substituteLoopStep(
+                      originalStepInput[key] =
+                        automationUtils.substituteLoopStep(
                           value,
                           `steps.${loopStepNumber}`
                         )
@@ -361,30 +401,42 @@ class Orchestrator {
                   index === env.AUTOMATION_MAX_ITERATIONS ||
                   index === parseInt(loopStep.inputs.iterations)
                 ) {
-            this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
+                  this.updateContextAndOutput(
+                    loopStepNumber,
+                    step,
+                    tempOutput,
+                    {
                       status: AutomationErrors.MAX_ITERATIONS,
                       success: true,
-            })
+                    }
+                  )
                   loopSteps = undefined
                   loopStep = undefined
                   break
                 }
 
                 let isFailure = false
-          const currentItem = this._context.steps[loopStepNumber]?.currentItem
+                const currentItem =
+                  this._context.steps[loopStepNumber]?.currentItem
                 if (currentItem && typeof currentItem === "object") {
                   isFailure = Object.keys(currentItem).some(value => {
                     return currentItem[value] === loopStep?.inputs.failure
                   })
                 } else {
-            isFailure = currentItem && currentItem === loopStep.inputs.failure
+                  isFailure =
+                    currentItem && currentItem === loopStep.inputs.failure
                 }
 
                 if (isFailure) {
-            this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
+                  this.updateContextAndOutput(
+                    loopStepNumber,
+                    step,
+                    tempOutput,
+                    {
                       status: AutomationErrors.FAILURE_CONDITION,
                       success: false,
-            })
+                    }
+                  )
                   loopSteps = undefined
                   loopStep = undefined
                   break
@@ -393,14 +445,22 @@ class Orchestrator {
 
               // execution stopped, record state for that
               if (stopped) {
-          this.updateExecutionOutput(step.id, step.stepId, {}, STOPPED_STATUS)
+                this.updateExecutionOutput(
+                  step.id,
+                  step.stepId,
+                  {},
+                  STOPPED_STATUS
+                )
                 continue
               }
 
               // If it's a loop step, we need to manually add the bindings to the context
               let stepFn = await this.getStepFunctionality(step.stepId)
               let inputs = await processObject(originalStepInput, this._context)
-        inputs = automationUtils.cleanInputValues(inputs, step.schema.inputs)
+              inputs = automationUtils.cleanInputValues(
+                inputs,
+                step.schema.inputs
+              )
 
               try {
                 // appId is always passed
@@ -416,10 +476,15 @@ class Orchestrator {
                 // so that we can finish iterating through the steps and record that it stopped
                 if (step.stepId === FILTER_STEP_ID && !outputs.result) {
                   stopped = true
-            this.updateExecutionOutput(step.id, step.stepId, step.inputs, {
+                  this.updateExecutionOutput(
+                    step.id,
+                    step.stepId,
+                    step.inputs,
+                    {
                       ...outputs,
                       ...STOPPED_STATUS,
-            })
+                    }
+                  )
                   continue
                 }
                 if (loopStep && loopSteps) {
@@ -446,6 +511,9 @@ class Orchestrator {
                 }
               }
             }
+          } finally {
+            stepSpan?.finish()
+          }
 
           if (loopStep && iterations === 0) {
             loopStep = undefined
@@ -515,6 +583,8 @@ class Orchestrator {
         }
         return this.executionOutput
       }
+    )
+  }
 }
 
 export function execute(job: Job<AutomationData>, callback: WorkerCallback) {