Add DataDog tracing to automations.
This commit is contained in:
Sam Rose
parent
654f807d33
commit
406410d916
|
|
@ -16,6 +16,7 @@ import {
|
||||||
} from "@budibase/types"
|
} from "@budibase/types"
|
||||||
import sdk from "../sdk"
|
import sdk from "../sdk"
|
||||||
import { automationsEnabled } from "../features"
|
import { automationsEnabled } from "../features"
|
||||||
|
import tracer from "dd-trace"
|
||||||
|
|
||||||
const REBOOT_CRON = "@reboot"
|
const REBOOT_CRON = "@reboot"
|
||||||
const WH_STEP_ID = definitions.WEBHOOK.stepId
|
const WH_STEP_ID = definitions.WEBHOOK.stepId
|
||||||
|
|
@ -39,26 +40,62 @@ function loggingArgs(job: AutomationJob) {
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function processEvent(job: AutomationJob) {
|
export async function processEvent(job: AutomationJob) {
|
||||||
const appId = job.data.event.appId!
|
return tracer.trace(
|
||||||
const automationId = job.data.automation._id!
|
"processEvent",
|
||||||
const task = async () => {
|
{ resource: "automation" },
|
||||||
try {
|
async span => {
|
||||||
// need to actually await these so that an error can be captured properly
|
const appId = job.data.event.appId!
|
||||||
console.log("automation running", ...loggingArgs(job))
|
const automationId = job.data.automation._id!
|
||||||
|
|
||||||
const runFn = () => Runner.run(job)
|
span?.addTags({
|
||||||
const result = await quotas.addAutomation(runFn, {
|
appId,
|
||||||
automationId,
|
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,
|
||||||
|
},
|
||||||
|
},
|
||||||
})
|
})
|
||||||
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 })
|
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)
|
||||||
|
const result = await quotas.addAutomation(runFn, {
|
||||||
|
automationId,
|
||||||
|
})
|
||||||
|
console.log("automation completed", ...loggingArgs(job))
|
||||||
|
return result
|
||||||
|
} catch (err) {
|
||||||
|
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(
|
export async function updateTestHistory(
|
||||||
|
|
|
||||||
|
|
@ -34,6 +34,7 @@ import { cloneDeep } from "lodash/fp"
|
||||||
import { performance } from "perf_hooks"
|
import { performance } from "perf_hooks"
|
||||||
import * as sdkUtils from "../sdk/utils"
|
import * as sdkUtils from "../sdk/utils"
|
||||||
import env from "../environment"
|
import env from "../environment"
|
||||||
|
import tracer from "dd-trace"
|
||||||
|
|
||||||
threadUtils.threadSetup()
|
threadUtils.threadSetup()
|
||||||
const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
|
const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
|
||||||
|
|
@ -242,278 +243,347 @@ class Orchestrator {
|
||||||
}
|
}
|
||||||
|
|
||||||
async execute(): Promise<any> {
|
async execute(): Promise<any> {
|
||||||
// this will retrieve from context created at start of thread
|
return tracer.trace(
|
||||||
this._context.env = await sdkUtils.getEnvironmentVariables()
|
"Orchestrator.execute",
|
||||||
let automation = this._automation
|
{ resource: "automation" },
|
||||||
let stopped = false
|
async span => {
|
||||||
let loopStep: AutomationStep | undefined = undefined
|
span?.addTags({
|
||||||
|
appId: this._appId,
|
||||||
|
automationId: this._automation._id,
|
||||||
|
})
|
||||||
|
|
||||||
let stepCount = 0
|
// this will retrieve from context created at start of thread
|
||||||
let loopStepNumber: any = undefined
|
this._context.env = await sdkUtils.getEnvironmentVariables()
|
||||||
let loopSteps: LoopStep[] | undefined = []
|
let automation = this._automation
|
||||||
let metadata
|
let stopped = false
|
||||||
let timeoutFlag = false
|
let loopStep: AutomationStep | undefined = undefined
|
||||||
let wasLoopStep = false
|
|
||||||
let timeout = this._job.data.event.timeout
|
|
||||||
// check if this is a recurring automation,
|
|
||||||
if (isProdAppID(this._appId) && isRecurring(automation)) {
|
|
||||||
metadata = await this.getMetadata()
|
|
||||||
const shouldStop = await this.checkIfShouldStop(metadata)
|
|
||||||
if (shouldStop) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
const start = performance.now()
|
|
||||||
for (let step of automation.definition.steps) {
|
|
||||||
if (timeoutFlag) {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
if (timeout) {
|
let stepCount = 0
|
||||||
setTimeout(() => {
|
let loopStepNumber: any = undefined
|
||||||
timeoutFlag = true
|
let loopSteps: LoopStep[] | undefined = []
|
||||||
}, timeout || 12000)
|
let metadata
|
||||||
}
|
let timeoutFlag = false
|
||||||
|
let wasLoopStep = false
|
||||||
stepCount++
|
let timeout = this._job.data.event.timeout
|
||||||
let input: any,
|
// check if this is a recurring automation,
|
||||||
iterations = 1,
|
if (isProdAppID(this._appId) && isRecurring(automation)) {
|
||||||
iterationCount = 0
|
span?.addTags({ recurring: true })
|
||||||
|
metadata = await this.getMetadata()
|
||||||
if (step.stepId === LOOP_STEP_ID) {
|
const shouldStop = await this.checkIfShouldStop(metadata)
|
||||||
loopStep = step
|
if (shouldStop) {
|
||||||
loopStepNumber = stepCount
|
span?.addTags({ shouldStop: true })
|
||||||
continue
|
return
|
||||||
}
|
|
||||||
|
|
||||||
if (loopStep) {
|
|
||||||
input = await processObject(loopStep.inputs, this._context)
|
|
||||||
iterations = getLoopIterations(loopStep as LoopStep)
|
|
||||||
}
|
|
||||||
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 }
|
|
||||||
try {
|
|
||||||
loopStep.inputs.binding = automationUtils.typecastForLooping(
|
|
||||||
loopStep as LoopStep,
|
|
||||||
loopStep.inputs as LoopInput
|
|
||||||
)
|
|
||||||
} catch (err) {
|
|
||||||
this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
|
|
||||||
status: AutomationErrors.INCORRECT_TYPE,
|
|
||||||
success: false,
|
|
||||||
})
|
|
||||||
loopSteps = undefined
|
|
||||||
loopStep = undefined
|
|
||||||
break
|
|
||||||
}
|
|
||||||
let item = []
|
|
||||||
if (
|
|
||||||
typeof loopStep.inputs.binding === "string" &&
|
|
||||||
loopStep.inputs.option === "String"
|
|
||||||
) {
|
|
||||||
item = automationUtils.stringSplit(loopStep.inputs.binding)
|
|
||||||
} else if (Array.isArray(loopStep.inputs.binding)) {
|
|
||||||
item = loopStep.inputs.binding
|
|
||||||
}
|
|
||||||
this._context.steps[loopStepNumber] = {
|
|
||||||
currentItem: item[index],
|
|
||||||
}
|
|
||||||
|
|
||||||
// The "Loop" binding in the front end is "fake", so replace it here so the context can understand it
|
|
||||||
// Pretty hacky because we need to account for the row object
|
|
||||||
for (let [key, value] of Object.entries(originalStepInput)) {
|
|
||||||
if (typeof value === "object") {
|
|
||||||
for (let [innerKey, innerValue] of Object.entries(
|
|
||||||
originalStepInput[key]
|
|
||||||
)) {
|
|
||||||
if (typeof innerValue === "string") {
|
|
||||||
originalStepInput[key][innerKey] =
|
|
||||||
automationUtils.substituteLoopStep(
|
|
||||||
innerValue,
|
|
||||||
`steps.${loopStepNumber}`
|
|
||||||
)
|
|
||||||
} else if (typeof value === "object") {
|
|
||||||
for (let [innerObject, innerValue] of Object.entries(
|
|
||||||
originalStepInput[key][innerKey]
|
|
||||||
)) {
|
|
||||||
originalStepInput[key][innerKey][innerObject] =
|
|
||||||
automationUtils.substituteLoopStep(
|
|
||||||
innerValue as string,
|
|
||||||
`steps.${loopStepNumber}`
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
if (typeof value === "string") {
|
|
||||||
originalStepInput[key] = automationUtils.substituteLoopStep(
|
|
||||||
value,
|
|
||||||
`steps.${loopStepNumber}`
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (
|
|
||||||
index === env.AUTOMATION_MAX_ITERATIONS ||
|
|
||||||
index === parseInt(loopStep.inputs.iterations)
|
|
||||||
) {
|
|
||||||
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
|
|
||||||
if (currentItem && typeof currentItem === "object") {
|
|
||||||
isFailure = Object.keys(currentItem).some(value => {
|
|
||||||
return currentItem[value] === loopStep?.inputs.failure
|
|
||||||
})
|
|
||||||
} else {
|
|
||||||
isFailure = currentItem && currentItem === loopStep.inputs.failure
|
|
||||||
}
|
|
||||||
|
|
||||||
if (isFailure) {
|
|
||||||
this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
|
|
||||||
status: AutomationErrors.FAILURE_CONDITION,
|
|
||||||
success: false,
|
|
||||||
})
|
|
||||||
loopSteps = undefined
|
|
||||||
loopStep = undefined
|
|
||||||
break
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
const start = performance.now()
|
||||||
// execution stopped, record state for that
|
for (let step of automation.definition.steps) {
|
||||||
if (stopped) {
|
const stepSpan = tracer.startSpan("Orchestrator.execute.step", {
|
||||||
this.updateExecutionOutput(step.id, step.stepId, {}, STOPPED_STATUS)
|
childOf: span,
|
||||||
continue
|
})
|
||||||
}
|
stepSpan.addTags({
|
||||||
|
resource: "automation",
|
||||||
// If it's a loop step, we need to manually add the bindings to the context
|
step: {
|
||||||
let stepFn = await this.getStepFunctionality(step.stepId)
|
stepId: step.stepId,
|
||||||
let inputs = await processObject(originalStepInput, this._context)
|
id: step.id,
|
||||||
inputs = automationUtils.cleanInputValues(inputs, step.schema.inputs)
|
name: step.name,
|
||||||
|
type: step.type,
|
||||||
try {
|
title: step.stepTitle,
|
||||||
// appId is always passed
|
internal: step.internal,
|
||||||
const outputs = await stepFn({
|
deprecated: step.deprecated,
|
||||||
inputs: inputs,
|
},
|
||||||
appId: this._appId,
|
|
||||||
emitter: this._emitter,
|
|
||||||
context: this._context,
|
|
||||||
})
|
})
|
||||||
|
|
||||||
this._context.steps[stepCount] = outputs
|
let input: any,
|
||||||
// if filter causes us to stop execution don't break the loop, set a var
|
iterations = 1,
|
||||||
// so that we can finish iterating through the steps and record that it stopped
|
iterationCount = 0
|
||||||
if (step.stepId === FILTER_STEP_ID && !outputs.result) {
|
|
||||||
stopped = true
|
|
||||||
this.updateExecutionOutput(step.id, step.stepId, step.inputs, {
|
|
||||||
...outputs,
|
|
||||||
...STOPPED_STATUS,
|
|
||||||
})
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if (loopStep && loopSteps) {
|
|
||||||
loopSteps.push(outputs)
|
|
||||||
} else {
|
|
||||||
this.updateExecutionOutput(
|
|
||||||
step.id,
|
|
||||||
step.stepId,
|
|
||||||
step.inputs,
|
|
||||||
outputs
|
|
||||||
)
|
|
||||||
}
|
|
||||||
} catch (err) {
|
|
||||||
console.error(`Automation error - ${step.stepId} - ${err}`)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
if (loopStep) {
|
try {
|
||||||
iterationCount++
|
if (timeoutFlag) {
|
||||||
if (index === iterations - 1) {
|
span?.addTags({ timedOut: true })
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
if (timeout) {
|
||||||
|
setTimeout(() => {
|
||||||
|
timeoutFlag = true
|
||||||
|
}, timeout || 12000)
|
||||||
|
}
|
||||||
|
|
||||||
|
stepCount++
|
||||||
|
if (step.stepId === LOOP_STEP_ID) {
|
||||||
|
loopStep = step
|
||||||
|
loopStepNumber = stepCount
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
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,
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
loopStep.inputs.binding = automationUtils.typecastForLooping(
|
||||||
|
loopStep as LoopStep,
|
||||||
|
loopStep.inputs as LoopInput
|
||||||
|
)
|
||||||
|
} catch (err) {
|
||||||
|
this.updateContextAndOutput(
|
||||||
|
loopStepNumber,
|
||||||
|
step,
|
||||||
|
tempOutput,
|
||||||
|
{
|
||||||
|
status: AutomationErrors.INCORRECT_TYPE,
|
||||||
|
success: false,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
loopSteps = undefined
|
||||||
|
loopStep = undefined
|
||||||
|
break
|
||||||
|
}
|
||||||
|
let item = []
|
||||||
|
if (
|
||||||
|
typeof loopStep.inputs.binding === "string" &&
|
||||||
|
loopStep.inputs.option === "String"
|
||||||
|
) {
|
||||||
|
item = automationUtils.stringSplit(loopStep.inputs.binding)
|
||||||
|
} else if (Array.isArray(loopStep.inputs.binding)) {
|
||||||
|
item = loopStep.inputs.binding
|
||||||
|
}
|
||||||
|
this._context.steps[loopStepNumber] = {
|
||||||
|
currentItem: item[index],
|
||||||
|
}
|
||||||
|
|
||||||
|
// The "Loop" binding in the front end is "fake", so replace it here so the context can understand it
|
||||||
|
// Pretty hacky because we need to account for the row object
|
||||||
|
for (let [key, value] of Object.entries(originalStepInput)) {
|
||||||
|
if (typeof value === "object") {
|
||||||
|
for (let [innerKey, innerValue] of Object.entries(
|
||||||
|
originalStepInput[key]
|
||||||
|
)) {
|
||||||
|
if (typeof innerValue === "string") {
|
||||||
|
originalStepInput[key][innerKey] =
|
||||||
|
automationUtils.substituteLoopStep(
|
||||||
|
innerValue,
|
||||||
|
`steps.${loopStepNumber}`
|
||||||
|
)
|
||||||
|
} else if (typeof value === "object") {
|
||||||
|
for (let [innerObject, innerValue] of Object.entries(
|
||||||
|
originalStepInput[key][innerKey]
|
||||||
|
)) {
|
||||||
|
originalStepInput[key][innerKey][innerObject] =
|
||||||
|
automationUtils.substituteLoopStep(
|
||||||
|
innerValue as string,
|
||||||
|
`steps.${loopStepNumber}`
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
if (typeof value === "string") {
|
||||||
|
originalStepInput[key] =
|
||||||
|
automationUtils.substituteLoopStep(
|
||||||
|
value,
|
||||||
|
`steps.${loopStepNumber}`
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (
|
||||||
|
index === env.AUTOMATION_MAX_ITERATIONS ||
|
||||||
|
index === parseInt(loopStep.inputs.iterations)
|
||||||
|
) {
|
||||||
|
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
|
||||||
|
if (currentItem && typeof currentItem === "object") {
|
||||||
|
isFailure = Object.keys(currentItem).some(value => {
|
||||||
|
return currentItem[value] === loopStep?.inputs.failure
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
isFailure =
|
||||||
|
currentItem && currentItem === loopStep.inputs.failure
|
||||||
|
}
|
||||||
|
|
||||||
|
if (isFailure) {
|
||||||
|
this.updateContextAndOutput(
|
||||||
|
loopStepNumber,
|
||||||
|
step,
|
||||||
|
tempOutput,
|
||||||
|
{
|
||||||
|
status: AutomationErrors.FAILURE_CONDITION,
|
||||||
|
success: false,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
loopSteps = undefined
|
||||||
|
loopStep = undefined
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// execution stopped, record state for that
|
||||||
|
if (stopped) {
|
||||||
|
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
|
||||||
|
)
|
||||||
|
|
||||||
|
try {
|
||||||
|
// appId is always passed
|
||||||
|
const outputs = await stepFn({
|
||||||
|
inputs: inputs,
|
||||||
|
appId: this._appId,
|
||||||
|
emitter: this._emitter,
|
||||||
|
context: this._context,
|
||||||
|
})
|
||||||
|
|
||||||
|
this._context.steps[stepCount] = outputs
|
||||||
|
// if filter causes us to stop execution don't break the loop, set a var
|
||||||
|
// 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,
|
||||||
|
{
|
||||||
|
...outputs,
|
||||||
|
...STOPPED_STATUS,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if (loopStep && loopSteps) {
|
||||||
|
loopSteps.push(outputs)
|
||||||
|
} else {
|
||||||
|
this.updateExecutionOutput(
|
||||||
|
step.id,
|
||||||
|
step.stepId,
|
||||||
|
step.inputs,
|
||||||
|
outputs
|
||||||
|
)
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
console.error(`Automation error - ${step.stepId} - ${err}`)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if (loopStep) {
|
||||||
|
iterationCount++
|
||||||
|
if (index === iterations - 1) {
|
||||||
|
loopStep = undefined
|
||||||
|
this._context.steps.splice(loopStepNumber, 1)
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
stepSpan?.finish()
|
||||||
|
}
|
||||||
|
|
||||||
|
if (loopStep && iterations === 0) {
|
||||||
loopStep = undefined
|
loopStep = undefined
|
||||||
|
this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
|
||||||
|
id: step.id,
|
||||||
|
stepId: step.stepId,
|
||||||
|
outputs: {
|
||||||
|
status: AutomationStepStatus.NO_ITERATIONS,
|
||||||
|
success: true,
|
||||||
|
},
|
||||||
|
inputs: {},
|
||||||
|
})
|
||||||
|
|
||||||
this._context.steps.splice(loopStepNumber, 1)
|
this._context.steps.splice(loopStepNumber, 1)
|
||||||
break
|
iterations = 1
|
||||||
|
}
|
||||||
|
|
||||||
|
// Delete the step after the loop step as it's irrelevant, since information is included
|
||||||
|
// in the loop step
|
||||||
|
if (wasLoopStep && !loopStep) {
|
||||||
|
this._context.steps.splice(loopStepNumber + 1, 1)
|
||||||
|
wasLoopStep = false
|
||||||
|
}
|
||||||
|
if (loopSteps && loopSteps.length) {
|
||||||
|
let tempOutput = {
|
||||||
|
success: true,
|
||||||
|
items: loopSteps,
|
||||||
|
iterations: iterationCount,
|
||||||
|
}
|
||||||
|
this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
|
||||||
|
id: step.id,
|
||||||
|
stepId: step.stepId,
|
||||||
|
outputs: tempOutput,
|
||||||
|
inputs: step.inputs,
|
||||||
|
})
|
||||||
|
this._context.steps[loopStepNumber] = tempOutput
|
||||||
|
|
||||||
|
wasLoopStep = true
|
||||||
|
loopSteps = []
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (loopStep && iterations === 0) {
|
const end = performance.now()
|
||||||
loopStep = undefined
|
const executionTime = end - start
|
||||||
this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
|
|
||||||
id: step.id,
|
|
||||||
stepId: step.stepId,
|
|
||||||
outputs: {
|
|
||||||
status: AutomationStepStatus.NO_ITERATIONS,
|
|
||||||
success: true,
|
|
||||||
},
|
|
||||||
inputs: {},
|
|
||||||
})
|
|
||||||
|
|
||||||
this._context.steps.splice(loopStepNumber, 1)
|
console.info(
|
||||||
iterations = 1
|
`Automation ID: ${automation._id} Execution time: ${executionTime} milliseconds`,
|
||||||
}
|
{
|
||||||
|
_logKey: "automation",
|
||||||
|
executionTime,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
// Delete the step after the loop step as it's irrelevant, since information is included
|
// store the logs for the automation run
|
||||||
// in the loop step
|
try {
|
||||||
if (wasLoopStep && !loopStep) {
|
await storeLog(this._automation, this.executionOutput)
|
||||||
this._context.steps.splice(loopStepNumber + 1, 1)
|
} catch (e: any) {
|
||||||
wasLoopStep = false
|
if (e.status === 413 && e.request?.data) {
|
||||||
}
|
// if content is too large we shouldn't log it
|
||||||
if (loopSteps && loopSteps.length) {
|
delete e.request.data
|
||||||
let tempOutput = {
|
e.request.data = { message: "removed due to large size" }
|
||||||
success: true,
|
}
|
||||||
items: loopSteps,
|
logging.logAlert("Error writing automation log", e)
|
||||||
iterations: iterationCount,
|
|
||||||
}
|
}
|
||||||
this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
|
if (isProdAppID(this._appId) && isRecurring(automation) && metadata) {
|
||||||
id: step.id,
|
await this.updateMetadata(metadata)
|
||||||
stepId: step.stepId,
|
}
|
||||||
outputs: tempOutput,
|
return this.executionOutput
|
||||||
inputs: step.inputs,
|
|
||||||
})
|
|
||||||
this._context.steps[loopStepNumber] = tempOutput
|
|
||||||
|
|
||||||
wasLoopStep = true
|
|
||||||
loopSteps = []
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
const end = performance.now()
|
|
||||||
const executionTime = end - start
|
|
||||||
|
|
||||||
console.info(
|
|
||||||
`Automation ID: ${automation._id} Execution time: ${executionTime} milliseconds`,
|
|
||||||
{
|
|
||||||
_logKey: "automation",
|
|
||||||
executionTime,
|
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
// store the logs for the automation run
|
|
||||||
try {
|
|
||||||
await storeLog(this._automation, this.executionOutput)
|
|
||||||
} catch (e: any) {
|
|
||||||
if (e.status === 413 && e.request?.data) {
|
|
||||||
// if content is too large we shouldn't log it
|
|
||||||
delete e.request.data
|
|
||||||
e.request.data = { message: "removed due to large size" }
|
|
||||||
}
|
|
||||||
logging.logAlert("Error writing automation log", e)
|
|
||||||
}
|
|
||||||
if (isProdAppID(this._appId) && isRecurring(automation) && metadata) {
|
|
||||||
await this.updateMetadata(metadata)
|
|
||||||
}
|
|
||||||
return this.executionOutput
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue