CodeMirror/budibase
1
0
Fork 0
mirror of synced 2024-07-05 14:31:17 +12:00
Code Issues Projects Releases Wiki Activity

Remove all custom tracing to see if it's the cause of the memory leak.

Browse source
This commit is contained in:
Sam Rose 2023-12-21 11:06:05 +00:00
parent 12a016221e
commit 3c0901f530
No known key found for this signature in database
9 changed files with 313 additions and 613 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
packages/backend-core/src/db/couch/DatabaseImpl.ts
View file

@ -17,7 +17,6 @@ import { directCouchUrlCall } from "./utils"
import { getPouchDB } from "./pouchDB"
import { WriteStream, ReadStream } from "fs"
import { newid } from "../../docIds/newid"
import { DDInstrumentedDatabase } from "../instrumentation"
function buildNano(couchInfo: { url: string; cookie: string }) {
return Nano({
@ -36,8 +35,7 @@ export function DatabaseWithConnection(
connection: string,
opts?: DatabaseOpts
) {
const db = new DatabaseImpl(dbName, opts, connection)
return new DDInstrumentedDatabase(db)
return new DatabaseImpl(dbName, opts, connection)
}
export class DatabaseImpl implements Database {

3
packages/backend-core/src/db/db.ts
View file

@ -1,9 +1,8 @@
import { directCouchQuery, DatabaseImpl } from "./couch"
import { CouchFindOptions, Database, DatabaseOpts } from "@budibase/types"
import { DDInstrumentedDatabase } from "./instrumentation"
export function getDB(dbName: string, opts?: DatabaseOpts): Database {
return new DDInstrumentedDatabase(new DatabaseImpl(dbName, opts))
return new DatabaseImpl(dbName, opts)
}
// we have to use a callback for this so that we can close

156
packages/backend-core/src/db/instrumentation.ts
View file

@ -1,156 +0,0 @@
import {
DocumentScope,
DocumentDestroyResponse,
DocumentInsertResponse,
DocumentBulkResponse,
OkResponse,
} from "@budibase/nano"
import {
AllDocsResponse,
AnyDocument,
Database,
DatabaseDumpOpts,
DatabasePutOpts,
DatabaseQueryOpts,
Document,
} from "@budibase/types"
import tracer from "dd-trace"
import { Writable } from "stream"
export class DDInstrumentedDatabase implements Database {
constructor(private readonly db: Database) {}
get name(): string {
return this.db.name
}
exists(): Promise<boolean> {
return tracer.trace("db.exists", span => {
span?.addTags({ db_name: this.name })
return this.db.exists()
})
}
checkSetup(): Promise<DocumentScope<any>> {
return tracer.trace("db.checkSetup", span => {
span?.addTags({ db_name: this.name })
return this.db.checkSetup()
})
}
get<T extends Document>(id?: string | undefined): Promise<T> {
return tracer.trace("db.get", span => {
span?.addTags({ db_name: this.name, doc_id: id })
return this.db.get(id)
})
}
getMultiple<T extends Document>(
ids: string[],
opts?: { allowMissing?: boolean | undefined } | undefined
): Promise<T[]> {
return tracer.trace("db.getMultiple", span => {
span?.addTags({
db_name: this.name,
num_docs: ids.length,
allow_missing: opts?.allowMissing,
})
return this.db.getMultiple(ids, opts)
})
}
remove(
id: string | Document,
rev?: string | undefined
): Promise<DocumentDestroyResponse> {
return tracer.trace("db.remove", span => {
span?.addTags({ db_name: this.name, doc_id: id })
return this.db.remove(id, rev)
})
}
put(
document: AnyDocument,
opts?: DatabasePutOpts | undefined
): Promise<DocumentInsertResponse> {
return tracer.trace("db.put", span => {
span?.addTags({ db_name: this.name, doc_id: document._id })
return this.db.put(document, opts)
})
}
bulkDocs(documents: AnyDocument[]): Promise<DocumentBulkResponse[]> {
return tracer.trace("db.bulkDocs", span => {
span?.addTags({ db_name: this.name, num_docs: documents.length })
return this.db.bulkDocs(documents)
})
}
allDocs<T extends Document>(
params: DatabaseQueryOpts
): Promise<AllDocsResponse<T>> {
return tracer.trace("db.allDocs", span => {
span?.addTags({ db_name: this.name })
return this.db.allDocs(params)
})
}
query<T extends Document>(
viewName: string,
params: DatabaseQueryOpts
): Promise<AllDocsResponse<T>> {
return tracer.trace("db.query", span => {
span?.addTags({ db_name: this.name, view_name: viewName })
return this.db.query(viewName, params)
})
}
destroy(): Promise<void | OkResponse> {
return tracer.trace("db.destroy", span => {
span?.addTags({ db_name: this.name })
return this.db.destroy()
})
}
compact(): Promise<void | OkResponse> {
return tracer.trace("db.compact", span => {
span?.addTags({ db_name: this.name })
return this.db.compact()
})
}
dump(stream: Writable, opts?: DatabaseDumpOpts | undefined): Promise<any> {
return tracer.trace("db.dump", span => {
span?.addTags({ db_name: this.name })
return this.db.dump(stream, opts)
})
}
load(...args: any[]): Promise<any> {
return tracer.trace("db.load", span => {
span?.addTags({ db_name: this.name })
return this.db.load(...args)
})
}
createIndex(...args: any[]): Promise<any> {
return tracer.trace("db.createIndex", span => {
span?.addTags({ db_name: this.name })
return this.db.createIndex(...args)
})
}
deleteIndex(...args: any[]): Promise<any> {
return tracer.trace("db.deleteIndex", span => {
span?.addTags({ db_name: this.name })
return this.db.deleteIndex(...args)
})
}
getIndexes(...args: any[]): Promise<any> {
return tracer.trace("db.getIndexes", span => {
span?.addTags({ db_name: this.name })
return this.db.getIndexes(...args)
})
}
}

6
packages/backend-core/src/logging/pino/logger.ts
View file

@ -5,7 +5,6 @@ import { IdentityType } from "@budibase/types"
import env from "../../environment"
import * as context from "../../context"
import * as correlation from "../correlation"
import tracer from "dd-trace"
import { formats } from "dd-trace/ext"
import { localFileDestination } from "../system"
@ -117,11 +116,6 @@ if (!env.DISABLE_PINO_LOGGER) {
correlationId: correlation.getId(),
}
const span = tracer.scope().active()
if (span) {
tracer.inject(span.context(), formats.LOG, contextObject)
}
const mergingObject: any = {
err: error,
pid: process.pid,

38
packages/server/src/automations/utils.ts
View file

@ -16,7 +16,6 @@ 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
@ -40,37 +39,9 @@ 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
@ -83,19 +54,12 @@ export async function processEvent(job: AutomationJob) {
console.log("automation completed", ...loggingArgs(job))
return result
} catch (err) {
span?.addTags({ error: true })
console.error(
`automation was unable to run`,
err,
...loggingArgs(job)
)
console.error(`automation was unable to run`, err, ...loggingArgs(job))
return { err }
}
}
return await context.doInAutomationContext({ appId, automationId, task })
}
)
}
export async function updateTestHistory(

10
packages/server/src/jsRunner.ts
View file

@ -2,13 +2,10 @@ import vm from "vm"
import env from "./environment"
import { setJSRunner } from "@budibase/string-templates"
import { context, timers } from "@budibase/backend-core"
import tracer from "dd-trace"
type TrackerFn = <T>(f: () => T) => T
export function init() {
setJSRunner((js: string, ctx: vm.Context) => {
return tracer.trace("runJS", {}, span => {
const perRequestLimit = env.JS_PER_REQUEST_TIME_LIMIT_MS
let track: TrackerFn = f => f()
if (perRequestLimit) {
@ -19,12 +16,6 @@ export function init() {
timers.ExecutionTimeTracker.withLimit(perRequestLimit)
}
track = bbCtx.jsExecutionTracker.track.bind(bbCtx.jsExecutionTracker)
span?.addTags({
js: {
limitMS: bbCtx.jsExecutionTracker.limitMs,
elapsedMS: bbCtx.jsExecutionTracker.elapsedMS,
},
})
}
}
@ -41,5 +32,4 @@ export function init() {
})
)
})
})
}

14
packages/server/src/middleware/currentapp.ts
View file

@ -12,7 +12,6 @@ import { getCachedSelf } from "../utilities/global"
import env from "../environment"
import { isWebhookEndpoint } from "./utils"
import { UserCtx, ContextUser } from "@budibase/types"
import tracer from "dd-trace"
export default async (ctx: UserCtx, next: any) => {
// try to get the appID from the request
@ -21,11 +20,6 @@ export default async (ctx: UserCtx, next: any) => {
return next()
}
if (requestAppId) {
const span = tracer.scope().active()
span?.setTag("appId", requestAppId)
}
// deny access to application preview
if (!env.isTest()) {
if (
@ -76,14 +70,6 @@ export default async (ctx: UserCtx, next: any) => {
return next()
}
if (ctx.user) {
const span = tracer.scope().active()
if (ctx.user._id) {
span?.setTag("userId", ctx.user._id)
}
span?.setTag("tenantId", ctx.user.tenantId)
}
const userId = ctx.user ? generateUserMetadataID(ctx.user._id!) : undefined
// if the user is not in the right tenant then make sure to wipe their cookie

93
packages/server/src/threads/automation.ts
View file

@ -34,7 +34,6 @@ 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
@ -243,15 +242,6 @@ 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
@ -267,39 +257,19 @@ 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
}
@ -319,7 +289,6 @@ 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)
@ -335,15 +304,10 @@ class Orchestrator {
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
@ -388,8 +352,7 @@ class Orchestrator {
}
} else {
if (typeof value === "string") {
originalStepInput[key] =
automationUtils.substituteLoopStep(
originalStepInput[key] = automationUtils.substituteLoopStep(
value,
`steps.${loopStepNumber}`
)
@ -401,42 +364,30 @@ 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
@ -445,22 +396,14 @@ 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
@ -476,15 +419,10 @@ 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) {
@ -511,9 +449,6 @@ class Orchestrator {
}
}
}
} finally {
stepSpan?.finish()
}
if (loopStep && iterations === 0) {
loopStep = undefined
@ -583,8 +518,6 @@ class Orchestrator {
}
return this.executionOutput
}
)
}
}
export function execute(job: Job<AutomationData>, callback: WorkerCallback) {

10
packages/server/src/utilities/rowProcessor/utils.ts
View file

@ -11,7 +11,6 @@ import {
Row,
Table,
} from "@budibase/types"
import tracer from "dd-trace"
interface FormulaOpts {
dynamic?: boolean
@ -51,9 +50,6 @@ export function processFormulas<T extends Row | Row[]>(
inputRows: T,
{ dynamic, contextRows }: FormulaOpts = { dynamic: true }
): T {
return tracer.trace("processFormulas", {}, span => {
const numRows = Array.isArray(inputRows) ? inputRows.length : 1
span?.addTags({ table_id: table._id, dynamic, numRows })
const rows = Array.isArray(inputRows) ? inputRows : [inputRows]
if (rows) {
for (let [column, schema] of Object.entries(table.schema)) {
@ -77,16 +73,12 @@ export function processFormulas<T extends Row | Row[]>(
let formula = schema.formula
rows[i] = {
...row,
[column]: tracer.trace("processStringSync", {}, span => {
span?.addTags({ table_id: table._id, column, static: isStatic })
return processStringSync(formula, context)
}),
[column]: processStringSync(formula, context),
}
}
}
}
return Array.isArray(inputRows) ? rows : rows[0]
})
}
/**