Revert "Merge pull request #11489 from Budibase/fix/automation-disable"

This reverts commit e6319cce67, reversing
changes made to 7c12280fe3.
This commit is contained in:
Rory Powell 2023-09-01 15:12:23 +01:00
parent fae3943ba8
commit 64af549341
10 changed files with 55 additions and 156 deletions

View File

@ -8,7 +8,6 @@ import {
DatabasePutOpts, DatabasePutOpts,
DatabaseCreateIndexOpts, DatabaseCreateIndexOpts,
DatabaseDeleteIndexOpts, DatabaseDeleteIndexOpts,
DocExistsResponse,
Document, Document,
isDocument, isDocument,
} from "@budibase/types" } from "@budibase/types"
@ -121,19 +120,6 @@ export class DatabaseImpl implements Database {
return this.updateOutput(() => db.get(id)) return this.updateOutput(() => db.get(id))
} }
async docExists(docId: string): Promise<DocExistsResponse> {
const db = await this.checkSetup()
let _rev, exists
try {
const { etag } = await db.head(docId)
_rev = etag
exists = true
} catch (err) {
exists = false
}
return { _rev, exists }
}
async remove(idOrDoc: string | Document, rev?: string) { async remove(idOrDoc: string | Document, rev?: string) {
const db = await this.checkSetup() const db = await this.checkSetup()
let _id: string let _id: string

View File

@ -39,8 +39,9 @@ import {
} from "../../db/defaultData/datasource_bb_default" } from "../../db/defaultData/datasource_bb_default"
import { removeAppFromUserRoles } from "../../utilities/workerRequests" import { removeAppFromUserRoles } from "../../utilities/workerRequests"
import { stringToReadStream } from "../../utilities" import { stringToReadStream } from "../../utilities"
import { doesUserHaveLock } from "../../utilities/redis" import { doesUserHaveLock, getLocksById } from "../../utilities/redis"
import { cleanupAutomations } from "../../automations/utils" import { cleanupAutomations } from "../../automations/utils"
import { checkAppMetadata } from "../../automations/logging"
import { getUniqueRows } from "../../utilities/usageQuota/rows" import { getUniqueRows } from "../../utilities/usageQuota/rows"
import { groups, licensing, quotas } from "@budibase/pro" import { groups, licensing, quotas } from "@budibase/pro"
import { import {
@ -50,6 +51,7 @@ import {
PlanType, PlanType,
Screen, Screen,
UserCtx, UserCtx,
ContextUser,
} from "@budibase/types" } from "@budibase/types"
import { BASE_LAYOUT_PROP_IDS } from "../../constants/layouts" import { BASE_LAYOUT_PROP_IDS } from "../../constants/layouts"
import sdk from "../../sdk" import sdk from "../../sdk"

View File

@ -20,7 +20,7 @@ import {
Automation, Automation,
AutomationActionStepId, AutomationActionStepId,
AutomationResults, AutomationResults,
Ctx, BBContext,
} from "@budibase/types" } from "@budibase/types"
import { getActionDefinitions as actionDefs } from "../../automations/actions" import { getActionDefinitions as actionDefs } from "../../automations/actions"
import sdk from "../../sdk" import sdk from "../../sdk"
@ -73,7 +73,7 @@ function cleanAutomationInputs(automation: Automation) {
return automation return automation
} }
export async function create(ctx: Ctx) { export async function create(ctx: BBContext) {
const db = context.getAppDB() const db = context.getAppDB()
let automation = ctx.request.body let automation = ctx.request.body
automation.appId = ctx.appId automation.appId = ctx.appId
@ -142,7 +142,7 @@ export async function handleStepEvents(
} }
} }
export async function update(ctx: Ctx) { export async function update(ctx: BBContext) {
const db = context.getAppDB() const db = context.getAppDB()
let automation = ctx.request.body let automation = ctx.request.body
automation.appId = ctx.appId automation.appId = ctx.appId
@ -193,7 +193,7 @@ export async function update(ctx: Ctx) {
builderSocket?.emitAutomationUpdate(ctx, automation) builderSocket?.emitAutomationUpdate(ctx, automation)
} }
export async function fetch(ctx: Ctx) { export async function fetch(ctx: BBContext) {
const db = context.getAppDB() const db = context.getAppDB()
const response = await db.allDocs( const response = await db.allDocs(
getAutomationParams(null, { getAutomationParams(null, {
@ -203,11 +203,12 @@ export async function fetch(ctx: Ctx) {
ctx.body = response.rows.map(row => row.doc) ctx.body = response.rows.map(row => row.doc)
} }
export async function find(ctx: Ctx) { export async function find(ctx: BBContext) {
ctx.body = await sdk.automations.get(ctx.params.id) const db = context.getAppDB()
ctx.body = await db.get(ctx.params.id)
} }
export async function destroy(ctx: Ctx) { export async function destroy(ctx: BBContext) {
const db = context.getAppDB() const db = context.getAppDB()
const automationId = ctx.params.id const automationId = ctx.params.id
const oldAutomation = await db.get<Automation>(automationId) const oldAutomation = await db.get<Automation>(automationId)
@ -221,11 +222,11 @@ export async function destroy(ctx: Ctx) {
builderSocket?.emitAutomationDeletion(ctx, automationId) builderSocket?.emitAutomationDeletion(ctx, automationId)
} }
export async function logSearch(ctx: Ctx) { export async function logSearch(ctx: BBContext) {
ctx.body = await automations.logs.logSearch(ctx.request.body) ctx.body = await automations.logs.logSearch(ctx.request.body)
} }
export async function clearLogError(ctx: Ctx) { export async function clearLogError(ctx: BBContext) {
const { automationId, appId } = ctx.request.body const { automationId, appId } = ctx.request.body
await context.doInAppContext(appId, async () => { await context.doInAppContext(appId, async () => {
const db = context.getProdAppDB() const db = context.getProdAppDB()
@ -244,15 +245,15 @@ export async function clearLogError(ctx: Ctx) {
}) })
} }
export async function getActionList(ctx: Ctx) { export async function getActionList(ctx: BBContext) {
ctx.body = await getActionDefinitions() ctx.body = await getActionDefinitions()
} }
export async function getTriggerList(ctx: Ctx) { export async function getTriggerList(ctx: BBContext) {
ctx.body = getTriggerDefinitions() ctx.body = getTriggerDefinitions()
} }
export async function getDefinitionList(ctx: Ctx) { export async function getDefinitionList(ctx: BBContext) {
ctx.body = { ctx.body = {
trigger: getTriggerDefinitions(), trigger: getTriggerDefinitions(),
action: await getActionDefinitions(), action: await getActionDefinitions(),
@ -265,7 +266,7 @@ export async function getDefinitionList(ctx: Ctx) {
* * * *
*********************/ *********************/
export async function trigger(ctx: Ctx) { export async function trigger(ctx: BBContext) {
const db = context.getAppDB() const db = context.getAppDB()
let automation = await db.get<Automation>(ctx.params.id) let automation = await db.get<Automation>(ctx.params.id)
@ -310,7 +311,7 @@ function prepareTestInput(input: any) {
return input return input
} }
export async function test(ctx: Ctx) { export async function test(ctx: BBContext) {
const db = context.getAppDB() const db = context.getAppDB()
let automation = await db.get<Automation>(ctx.params.id) let automation = await db.get<Automation>(ctx.params.id)
await setTestFlag(automation._id!) await setTestFlag(automation._id!)

View File

@ -6,11 +6,11 @@ import { isDevAppID } from "../db/utils"
// need this to call directly, so we can get a response // need this to call directly, so we can get a response
import { automationQueue } from "./bullboard" import { automationQueue } from "./bullboard"
import { checkTestFlag } from "../utilities/redis" import { checkTestFlag } from "../utilities/redis"
import * as utils from "./utils"
import env from "../environment" import env from "../environment"
import { context, db as dbCore } from "@budibase/backend-core" import { context, db as dbCore } from "@budibase/backend-core"
import { Automation, Row, AutomationData, AutomationJob } from "@budibase/types" import { Automation, Row, AutomationData, AutomationJob } from "@budibase/types"
import { executeSynchronously } from "../threads/automation" import { executeSynchronously } from "../threads/automation"
import sdk from "../sdk"
export const TRIGGER_DEFINITIONS = definitions export const TRIGGER_DEFINITIONS = definitions
const JOB_OPTS = { const JOB_OPTS = {
@ -142,7 +142,7 @@ export async function rebootTrigger() {
let automations = await getAllAutomations() let automations = await getAllAutomations()
let rebootEvents = [] let rebootEvents = []
for (let automation of automations) { for (let automation of automations) {
if (sdk.automations.isReboot(automation)) { if (utils.isRebootTrigger(automation)) {
const job = { const job = {
automation, automation,
event: { event: {

View File

@ -16,14 +16,13 @@ import {
} from "@budibase/types" } from "@budibase/types"
import sdk from "../sdk" import sdk from "../sdk"
const REBOOT_CRON = "@reboot"
const WH_STEP_ID = definitions.WEBHOOK.stepId const WH_STEP_ID = definitions.WEBHOOK.stepId
const CRON_STEP_ID = definitions.CRON.stepId
const Runner = new Thread(ThreadType.AUTOMATION) const Runner = new Thread(ThreadType.AUTOMATION)
function loggingArgs( function loggingArgs(job: AutomationJob) {
job: AutomationJob, return [
timing?: { start: number; complete?: boolean }
) {
const logs: any[] = [
{ {
_logKey: "automation", _logKey: "automation",
trigger: job.data.automation.definition.trigger.event, trigger: job.data.automation.definition.trigger.event,
@ -33,53 +32,24 @@ function loggingArgs(
jobId: job.id, jobId: job.id,
}, },
] ]
if (timing?.start) {
logs.push({
_logKey: "startTime",
start: timing.start,
})
}
if (timing?.start && timing?.complete) {
const end = new Date().getTime()
const duration = end - timing.start
logs.push({
_logKey: "endTime",
end,
})
logs.push({
_logKey: "duration",
duration,
})
}
return logs
} }
export async function processEvent(job: AutomationJob) { export async function processEvent(job: AutomationJob) {
const appId = job.data.event.appId! const appId = job.data.event.appId!
const automationId = job.data.automation._id! const automationId = job.data.automation._id!
const start = new Date().getTime()
const task = async () => { const task = async () => {
try { try {
// need to actually await these so that an error can be captured properly // need to actually await these so that an error can be captured properly
console.log("automation running", ...loggingArgs(job, { start })) console.log("automation running", ...loggingArgs(job))
const runFn = () => Runner.run(job) const runFn = () => Runner.run(job)
const result = await quotas.addAutomation(runFn, { const result = await quotas.addAutomation(runFn, {
automationId, automationId,
}) })
const end = new Date().getTime() console.log("automation completed", ...loggingArgs(job))
const duration = end - start
console.log(
"automation completed",
...loggingArgs(job, { start, complete: true })
)
return result return result
} catch (err) { } catch (err) {
console.error( console.error(`automation was unable to run`, err, ...loggingArgs(job))
`automation was unable to run`,
err,
...loggingArgs(job, { start, complete: true })
)
return { err } return { err }
} }
} }
@ -158,6 +128,19 @@ export async function clearMetadata() {
await db.bulkDocs(automationMetadata) await db.bulkDocs(automationMetadata)
} }
export function isCronTrigger(auto: Automation) {
return (
auto &&
auto.definition.trigger &&
auto.definition.trigger.stepId === CRON_STEP_ID
)
}
export function isRebootTrigger(auto: Automation) {
const trigger = auto ? auto.definition.trigger : null
return isCronTrigger(auto) && trigger?.inputs.cron === REBOOT_CRON
}
/** /**
* This function handles checking of any cron jobs that need to be enabled/updated. * This function handles checking of any cron jobs that need to be enabled/updated.
* @param {string} appId The ID of the app in which we are checking for webhooks * @param {string} appId The ID of the app in which we are checking for webhooks
@ -165,13 +148,13 @@ export async function clearMetadata() {
*/ */
export async function enableCronTrigger(appId: any, automation: Automation) { export async function enableCronTrigger(appId: any, automation: Automation) {
const trigger = automation ? automation.definition.trigger : null const trigger = automation ? automation.definition.trigger : null
const validCron = sdk.automations.isCron(automation) && trigger?.inputs.cron
const needsCreated =
!sdk.automations.isReboot(automation) &&
!sdk.automations.disabled(automation)
// need to create cron job // need to create cron job
if (validCron && needsCreated) { if (
isCronTrigger(automation) &&
!isRebootTrigger(automation) &&
trigger?.inputs.cron
) {
// make a job id rather than letting Bull decide, makes it easier to handle on way out // make a job id rather than letting Bull decide, makes it easier to handle on way out
const jobId = `${appId}_cron_${newid()}` const jobId = `${appId}_cron_${newid()}`
const job: any = await automationQueue.add( const job: any = await automationQueue.add(

View File

@ -1,38 +0,0 @@
import { context } from "@budibase/backend-core"
import { Automation, AutomationState, DocumentType } from "@budibase/types"
import { definitions } from "../../../automations/triggerInfo"
const REBOOT_CRON = "@reboot"
export async function exists(automationId: string) {
if (!automationId?.startsWith(DocumentType.AUTOMATION)) {
throw new Error("Invalid automation ID.")
}
const db = context.getAppDB()
return db.docExists(automationId)
}
export async function get(automationId: string) {
const db = context.getAppDB()
return (await db.get(automationId)) as Automation
}
export function disabled(automation: Automation) {
return automation.state === AutomationState.DISABLED || !hasSteps(automation)
}
export function isCron(automation: Automation) {
return (
automation?.definition.trigger &&
automation?.definition.trigger.stepId === definitions.CRON.stepId
)
}
export function isReboot(automation: Automation) {
const trigger = automation?.definition.trigger
return isCron(automation) && trigger?.inputs.cron === REBOOT_CRON
}
export function hasSteps(automation: Automation) {
return automation?.definition?.steps?.length > 0
}

View File

@ -1,9 +1,7 @@
import * as webhook from "./webhook" import * as webhook from "./webhook"
import * as utils from "./utils" import * as utils from "./utils"
import * as automations from "./automations"
export default { export default {
webhook, webhook,
utils, utils,
...automations,
} }

View File

@ -2,9 +2,9 @@ import { default as threadUtils } from "./utils"
import { Job } from "bull" import { Job } from "bull"
threadUtils.threadSetup() threadUtils.threadSetup()
import { import {
isRecurring,
disableCronById, disableCronById,
isErrorInOutput, isErrorInOutput,
isRecurring,
} from "../automations/utils" } from "../automations/utils"
import * as actions from "../automations/actions" import * as actions from "../automations/actions"
import * as automationUtils from "../automations/automationUtils" import * as automationUtils from "../automations/automationUtils"
@ -15,17 +15,17 @@ import { AutomationErrors, MAX_AUTOMATION_RECURRING_ERRORS } from "../constants"
import { storeLog } from "../automations/logging" import { storeLog } from "../automations/logging"
import { import {
Automation, Automation,
AutomationData,
AutomationJob,
AutomationMetadata,
AutomationStatus,
AutomationStep, AutomationStep,
AutomationStatus,
AutomationMetadata,
AutomationJob,
AutomationData,
} from "@budibase/types" } from "@budibase/types"
import { import {
AutomationContext,
LoopInput,
LoopStep, LoopStep,
LoopInput,
TriggerOutput, TriggerOutput,
AutomationContext,
} from "../definitions/automations" } from "../definitions/automations"
import { WorkerCallback } from "./definitions" import { WorkerCallback } from "./definitions"
import { context, logging } from "@budibase/backend-core" import { context, logging } from "@budibase/backend-core"
@ -34,8 +34,6 @@ 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 sdk from "../sdk"
const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
const LOOP_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.LOOP.stepId const LOOP_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.LOOP.stepId
const CRON_STEP_ID = triggerDefs.CRON.stepId const CRON_STEP_ID = triggerDefs.CRON.stepId
@ -516,8 +514,7 @@ class Orchestrator {
export function execute(job: Job<AutomationData>, callback: WorkerCallback) { export function execute(job: Job<AutomationData>, callback: WorkerCallback) {
const appId = job.data.event.appId const appId = job.data.event.appId
const automation = job.data.automation const automationId = job.data.automation._id
const automationId = automation._id
if (!appId) { if (!appId) {
throw new Error("Unable to execute, event doesn't contain app ID.") throw new Error("Unable to execute, event doesn't contain app ID.")
} }
@ -528,30 +525,10 @@ export function execute(job: Job<AutomationData>, callback: WorkerCallback) {
appId, appId,
automationId, automationId,
task: async () => { task: async () => {
let automation = job.data.automation,
isCron = sdk.automations.isCron(job.data.automation),
notFound = false
try {
automation = await sdk.automations.get(automationId)
} catch (err: any) {
// automation no longer exists
notFound = err
}
const disabled = sdk.automations.disabled(automation)
const stopAutomation = disabled || notFound
const envVars = await sdkUtils.getEnvironmentVariables() const envVars = await sdkUtils.getEnvironmentVariables()
// put into automation thread for whole context // put into automation thread for whole context
await context.doInEnvironmentContext(envVars, async () => { await context.doInEnvironmentContext(envVars, async () => {
const automationOrchestrator = new Orchestrator(job) const automationOrchestrator = new Orchestrator(job)
// hard stop on automations
if (isCron && stopAutomation) {
await automationOrchestrator.stopCron(
disabled ? "disabled" : "not_found"
)
}
if (stopAutomation) {
return
}
try { try {
const response = await automationOrchestrator.execute() const response = await automationOrchestrator.execute()
callback(null, response) callback(null, response)
@ -580,10 +557,11 @@ export function executeSynchronously(job: Job) {
// put into automation thread for whole context // put into automation thread for whole context
return context.doInEnvironmentContext(envVars, async () => { return context.doInEnvironmentContext(envVars, async () => {
const automationOrchestrator = new Orchestrator(job) const automationOrchestrator = new Orchestrator(job)
return await Promise.race([ const response = await Promise.race([
automationOrchestrator.execute(), automationOrchestrator.execute(),
timeoutPromise, timeoutPromise,
]) ])
return response
}) })
}) })
} }

View File

@ -100,10 +100,6 @@ export const AutomationStepIdArray = [
...Object.values(AutomationTriggerStepId), ...Object.values(AutomationTriggerStepId),
] ]
export enum AutomationState {
DISABLED = "disabled",
}
export interface Automation extends Document { export interface Automation extends Document {
definition: { definition: {
steps: AutomationStep[] steps: AutomationStep[]
@ -116,7 +112,6 @@ export interface Automation extends Document {
name: string name: string
internal?: boolean internal?: boolean
type?: string type?: string
state?: AutomationState
} }
interface BaseIOStructure { interface BaseIOStructure {

View File

@ -40,11 +40,6 @@ export type DatabasePutOpts = {
force?: boolean force?: boolean
} }
export type DocExistsResponse = {
_rev?: string
exists: boolean
}
export type DatabaseCreateIndexOpts = { export type DatabaseCreateIndexOpts = {
index: { index: {
fields: string[] fields: string[]
@ -95,7 +90,6 @@ export interface Database {
exists(): Promise<boolean> exists(): Promise<boolean>
checkSetup(): Promise<Nano.DocumentScope<any>> checkSetup(): Promise<Nano.DocumentScope<any>>
get<T>(id?: string): Promise<T> get<T>(id?: string): Promise<T>
docExists(id: string): Promise<DocExistsResponse>
remove( remove(
id: string | Document, id: string | Document,
rev?: string rev?: string