Merge branch 'master' into run-helm-lint-in-ci
This commit is contained in:
Sam Rose
GitHub
commit
01df5a7de2
|
|
@ -15,6 +15,7 @@ function newJob(queue: string, message: any) {
|
|||
timestamp: Date.now(),
|
||||
queue: queue,
|
||||
data: message,
|
||||
opts: {},
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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,12 +83,19 @@ 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(
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue