From fbccfbc7f50bad85ef90b37b9a5602a30ec97bae Mon Sep 17 00:00:00 2001 From: Declan Carroll Date: Mon, 16 Mar 2026 15:29:50 +0000 Subject: [PATCH] test(benchmark): Add Kafka and webhook benchmark framework (no-changelog) (#26761) Co-authored-by: Claude Opus 4.6 --- .../test-e2e-infrastructure-reusable.yml | 35 +++ .github/workflows/test-e2e-reusable.yml | 7 +- packages/testing/containers/services/kafka.ts | 60 +++++ packages/testing/containers/services/n8n.ts | 30 ++- .../containers/services/postgres-exporter.ts | 74 ++++++ .../testing/containers/services/postgres.ts | 2 + .../testing/containers/services/registry.ts | 2 + packages/testing/containers/services/types.ts | 2 + .../containers/services/victoria-metrics.ts | 16 +- packages/testing/containers/stack.ts | 2 + .../testing/containers/test-containers.ts | 2 + packages/testing/playwright/package.json | 9 +- .../testing/playwright/playwright-projects.ts | 78 ++++++ .../testing/playwright/playwright.config.ts | 8 +- .../reporters/benchmark-summary-reporter.ts | 205 ++++++++++++++++ .../benchmarks/harness/load-harness.ts | 148 ++++++++++++ .../benchmarks/harness/throughput-harness.ts | 187 +++++++++++++++ .../harness/webhook-throughput-harness.ts | 174 ++++++++++++++ .../infrastructure/benchmarks/kafka/README.md | 131 ++++++++++ .../kafka/load-30n-10kb-steady-200.spec.ts | 29 +++ .../kafka/load-30n-10kb-steady-300.spec.ts | 29 +++ .../kafka/load-30n-10kb-steady.spec.ts | 29 +++ .../kafka/load-60n-1kb-burst.spec.ts | 32 +++ .../kafka/throughput-10n-100kb.spec.ts | 47 ++++ .../kafka/throughput-10n-10kb.spec.ts | 42 ++++ .../kafka/throughput-30n-10kb.spec.ts | 42 ++++ .../kafka/throughput-60n-10kb.spec.ts | 42 ++++ .../webhook/throughput-async.spec.ts | 46 ++++ .../webhook/throughput-sync.spec.ts | 46 ++++ .../playwright/utils/benchmark/diagnostics.ts | 140 +++++++++++ .../utils/benchmark/execution-sampler.ts | 101 ++++++++ .../playwright/utils/benchmark/index.ts | 7 + .../utils/benchmark/kafka-driver.ts | 222 +++++++++++++++++ .../utils/benchmark/throughput-measure.ts | 227 ++++++++++++++++++ .../playwright/utils/benchmark/types.ts | 119 +++++++++ .../utils/benchmark/webhook-driver.ts | 66 +++++ .../utils/benchmark/workflow-builder.ts | 64 +++++ pnpm-lock.yaml | 179 ++++++++++++-- 38 files changed, 2643 insertions(+), 38 deletions(-) create mode 100644 .github/workflows/test-e2e-infrastructure-reusable.yml create mode 100644 packages/testing/containers/services/postgres-exporter.ts create mode 100644 packages/testing/playwright/reporters/benchmark-summary-reporter.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/harness/load-harness.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/harness/throughput-harness.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/harness/webhook-throughput-harness.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/README.md create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-200.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-300.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-60n-1kb-burst.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-100kb.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-10kb.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-30n-10kb.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-60n-10kb.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-async.spec.ts create mode 100644 packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-sync.spec.ts create mode 100644 packages/testing/playwright/utils/benchmark/diagnostics.ts create mode 100644 packages/testing/playwright/utils/benchmark/execution-sampler.ts create mode 100644 packages/testing/playwright/utils/benchmark/index.ts create mode 100644 packages/testing/playwright/utils/benchmark/kafka-driver.ts create mode 100644 packages/testing/playwright/utils/benchmark/throughput-measure.ts create mode 100644 packages/testing/playwright/utils/benchmark/types.ts create mode 100644 packages/testing/playwright/utils/benchmark/webhook-driver.ts create mode 100644 packages/testing/playwright/utils/benchmark/workflow-builder.ts diff --git a/.github/workflows/test-e2e-infrastructure-reusable.yml b/.github/workflows/test-e2e-infrastructure-reusable.yml new file mode 100644 index 00000000000..f6191402f2e --- /dev/null +++ b/.github/workflows/test-e2e-infrastructure-reusable.yml @@ -0,0 +1,35 @@ +name: 'Test: E2E Infrastructure' + +on: + workflow_call: + workflow_dispatch: + pull_request: + paths: + - 'packages/testing/playwright/tests/infrastructure/**' + - 'packages/testing/playwright/utils/benchmark/**' + - 'packages/testing/playwright/utils/performance-helper.ts' + - 'packages/testing/playwright/reporters/benchmark-summary-reporter.ts' + - 'packages/testing/containers/services/**' + - '.github/workflows/test-e2e-infrastructure-reusable.yml' + +jobs: + benchmark: + name: ${{ matrix.profile }} + strategy: + fail-fast: false + matrix: + include: + - profile: benchmark-direct + runner: blacksmith-4vcpu-ubuntu-2204 + - profile: benchmark-queue + runner: blacksmith-8vcpu-ubuntu-2204 + - profile: benchmark-queue-tuned + runner: blacksmith-8vcpu-ubuntu-2204 + uses: ./.github/workflows/test-e2e-reusable.yml + with: + test-mode: docker-build + test-command: pnpm --filter=n8n-playwright test:all --project='${{ matrix.profile }}:infrastructure' --workers=1 + shards: 1 + runner: ${{ matrix.runner }} + timeout-minutes: 60 + secrets: inherit diff --git a/.github/workflows/test-e2e-reusable.yml b/.github/workflows/test-e2e-reusable.yml index 43cdcfd7ed5..f468338d1fa 100644 --- a/.github/workflows/test-e2e-reusable.yml +++ b/.github/workflows/test-e2e-reusable.yml @@ -42,6 +42,11 @@ on: required: false default: false type: boolean + timeout-minutes: + description: 'Job timeout in minutes' + required: false + default: 30 + type: number upload-failure-artifacts: description: 'Upload test failure artifacts (screenshots, traces, videos). Enable for community PRs without Currents access.' required: false @@ -113,7 +118,7 @@ jobs: needs: matrix if: ${{ !cancelled() }} runs-on: ${{ vars.RUNNER_PROVIDER == 'github' && 'ubuntu-latest' || inputs.runner }} - timeout-minutes: 30 + timeout-minutes: ${{ inputs.timeout-minutes }} permissions: packages: read contents: read diff --git a/packages/testing/containers/services/kafka.ts b/packages/testing/containers/services/kafka.ts index 0283bcca008..e1ea5c4a0f3 100644 --- a/packages/testing/containers/services/kafka.ts +++ b/packages/testing/containers/services/kafka.ts @@ -49,6 +49,10 @@ export const kafka: Service = { }, }; +/** + * Test helper for interacting with a Kafka broker. + * Provides topic management, message publishing, consumer group monitoring, and consumption. + */ export class KafkaHelper { private readonly kafka: Kafka; @@ -61,6 +65,7 @@ export class KafkaHelper { }); } + /** Creates a topic with the given number of partitions. */ async createTopic(topic: string, numPartitions = 1): Promise { const admin = this.kafka.admin(); try { @@ -73,6 +78,7 @@ export class KafkaHelper { } } + /** Polls until a consumer group reaches 'Stable' state with active members, or times out. */ async waitForConsumerGroup( groupId: string, options: { timeoutMs?: number; pollIntervalMs?: number } = {}, @@ -101,6 +107,7 @@ export class KafkaHelper { } } + /** Publishes a single message to a topic. Lazily initializes the producer on first call. */ async publish(topic: string, message: string | object, key?: string): Promise { if (!this.producer) { this.producer = this.kafka.producer(); @@ -115,6 +122,59 @@ export class KafkaHelper { }); } + /** + * Publishes messages in chunked batches to stay under Kafka's message.max.bytes limit. + * Default batch size is 1000 messages; callers can override for large payloads. + */ + async publishBatch( + topic: string, + messages: Array<{ value: string | object; key?: string }>, + options: { batchSize?: number } = {}, + ): Promise { + if (!this.producer) { + this.producer = this.kafka.producer(); + await this.producer.connect(); + } + + const batchSize = Math.max(1, options.batchSize ?? 1000); + const kafkaMessages = messages.map((m) => ({ + key: m.key, + value: typeof m.value === 'string' ? m.value : JSON.stringify(m.value), + })); + + for (let i = 0; i < kafkaMessages.length; i += batchSize) { + const chunk = kafkaMessages.slice(i, i + batchSize); + await this.producer.send({ topic, messages: chunk }); + } + } + + /** Returns per-partition and total lag for a consumer group on a topic. */ + async getConsumerGroupLag( + groupId: string, + topic: string, + ): Promise<{ totalLag: number; partitions: Array<{ partition: number; lag: number }> }> { + const admin = this.kafka.admin(); + try { + await admin.connect(); + const offsets = await admin.fetchOffsets({ groupId, topics: [topic] }); + const topicOffsets = await admin.fetchTopicOffsets(topic); + + const consumerPartitions = offsets.find((o) => o.topic === topic)?.partitions ?? []; + const committedByPartition = new Map(consumerPartitions.map((p) => [p.partition, p.offset])); + const partitions = topicOffsets.map((tp) => { + const committedOffset = committedByPartition.get(tp.partition) ?? '0'; + const lag = parseInt(tp.high, 10) - parseInt(committedOffset, 10); + return { partition: tp.partition, lag }; + }); + const totalLag = partitions.reduce((sum, p) => sum + p.lag, 0); + + return { totalLag, partitions }; + } finally { + await admin.disconnect(); + } + } + + /** Consumes up to maxMessages from a topic, returning within timeoutMs. Used for test assertions. */ async consume( topic: string, options: { diff --git a/packages/testing/containers/services/n8n.ts b/packages/testing/containers/services/n8n.ts index 9d5d5954301..6588a1facf2 100644 --- a/packages/testing/containers/services/n8n.ts +++ b/packages/testing/containers/services/n8n.ts @@ -51,6 +51,7 @@ export interface N8NInstancesOptions { baseUrl?: string; allocatedPort?: number; resourceQuota?: { memory?: number; cpu?: number }; + workerResourceQuota?: { memory?: number; cpu?: number }; filesToMount?: FileToMount[]; } @@ -183,14 +184,22 @@ async function createContainer( export async function createN8NInstances( options: N8NInstancesOptions, ): Promise { - const { mains, workers, projectName, network, allocatedPort, resourceQuota, filesToMount } = - options; + const { + mains, + workers, + projectName, + network, + allocatedPort, + resourceQuota, + workerResourceQuota, + filesToMount, + } = options; const log = createElapsedLogger('n8n-instances'); const environment = computeEnvironment(options); const containers: StartedTestContainer[] = []; - const shared: SharedConfig = { + const mainShared: SharedConfig = { projectName, environment, network, @@ -198,6 +207,14 @@ export async function createN8NInstances( filesToMount, }; + const workerShared: SharedConfig = { + projectName, + environment, + network, + resourceQuota: workerResourceQuota ?? resourceQuota, + filesToMount, + }; + const instances: InstanceConfig[] = [ ...Array.from({ length: mains }, (_, i) => { const num = i + 1; @@ -226,7 +243,7 @@ export async function createN8NInstances( // Start main 1 first (handles DB migrations/setup) const [main1, ...remaining] = instances; log(`Starting main 1: ${main1.name} (DB setup)`); - containers.push(await createContainer(main1, shared)); + containers.push(await createContainer(main1, mainShared)); log('main 1 ready'); // Start remaining instances in parallel @@ -236,7 +253,10 @@ export async function createN8NInstances( remaining.map(async (instance) => { const type = instance.isWorker ? 'worker' : 'main'; log(`Starting ${type} ${instance.instanceNumber}: ${instance.name}`); - const container = await createContainer(instance, shared); + const container = await createContainer( + instance, + instance.isWorker ? workerShared : mainShared, + ); log(`${type} ${instance.instanceNumber} ready`); return container; }), diff --git a/packages/testing/containers/services/postgres-exporter.ts b/packages/testing/containers/services/postgres-exporter.ts new file mode 100644 index 00000000000..f357fc93792 --- /dev/null +++ b/packages/testing/containers/services/postgres-exporter.ts @@ -0,0 +1,74 @@ +import { GenericContainer, Wait } from 'testcontainers'; +import type { StartedNetwork } from 'testcontainers'; + +import { TEST_CONTAINER_IMAGES } from '../test-containers'; +import type { PostgresResult } from './postgres'; +import type { Service, ServiceResult, StartContext } from './types'; + +const HOSTNAME = 'postgres-exporter'; +export const EXPORTER_PORT = 9187; + +export interface PostgresExporterMeta { + host: string; + port: number; +} + +export type PostgresExporterResult = ServiceResult; + +/** + * Runs a Prometheus-compatible exporter that scrapes PostgreSQL internal statistics + * (connections, transactions, replication lag, etc.) and exposes them as metrics on /metrics. + * VictoriaMetrics scrapes this endpoint to make Postgres performance data queryable via PromQL. + * Auto-starts when both postgres and victoriaMetrics services are in use. + */ +export const postgresExporter: Service = { + description: 'Postgres Exporter', + dependsOn: ['postgres'], + + shouldStart(ctx: StartContext): boolean { + // Auto-start when both postgres and victoriaMetrics are in use + const services = ctx.config.services ?? []; + return ctx.usePostgres && services.includes('victoriaMetrics'); + }, + + async start( + network: StartedNetwork, + projectName: string, + _options?: unknown, + ctx?: StartContext, + ): Promise { + const pgResult = ctx?.serviceResults.postgres as PostgresResult | undefined; + if (!pgResult) { + throw new Error('Postgres service must start before postgres-exporter'); + } + + const { username, password, database } = pgResult.meta; + const dsn = `postgresql://${username}:${password}@postgres:5432/${database}?sslmode=disable`; + + const container = await new GenericContainer(TEST_CONTAINER_IMAGES.postgresExporter) + .withName(`${projectName}-${HOSTNAME}`) + .withNetwork(network) + .withNetworkAliases(HOSTNAME) + .withLabels({ + 'com.docker.compose.project': projectName, + 'com.docker.compose.service': HOSTNAME, + }) + .withEnvironment({ + DATA_SOURCE_NAME: dsn, + }) + .withExposedPorts(EXPORTER_PORT) + .withWaitStrategy( + Wait.forHttp('/metrics', EXPORTER_PORT).forStatusCode(200).withStartupTimeout(30000), + ) + .withReuse() + .start(); + + return { + container, + meta: { + host: HOSTNAME, + port: EXPORTER_PORT, + }, + }; + }, +}; diff --git a/packages/testing/containers/services/postgres.ts b/packages/testing/containers/services/postgres.ts index f0cf80a9581..00e021fa54f 100644 --- a/packages/testing/containers/services/postgres.ts +++ b/packages/testing/containers/services/postgres.ts @@ -41,6 +41,8 @@ export const postgres: Service = { 'synchronous_commit=off', '-c', 'full_page_writes=off', + '-c', + 'max_connections=200', ]) .withReuse() .start(); diff --git a/packages/testing/containers/services/registry.ts b/packages/testing/containers/services/registry.ts index 87a463d6317..977cf90452f 100644 --- a/packages/testing/containers/services/registry.ts +++ b/packages/testing/containers/services/registry.ts @@ -10,6 +10,7 @@ import { mysqlService } from './mysql'; import { ngrok } from './ngrok'; import { createObservabilityHelper } from './observability'; import { postgres } from './postgres'; +import { postgresExporter } from './postgres-exporter'; import { proxy, createProxyHelper } from './proxy'; import { redis } from './redis'; import { taskRunner } from './task-runner'; @@ -39,6 +40,7 @@ export const services: Record> = { mysql: mysqlService, localstack, kent, + postgresExporter, }; export const helperFactories: Partial = { diff --git a/packages/testing/containers/services/types.ts b/packages/testing/containers/services/types.ts index 4cce3e66838..e28c0c2edf2 100644 --- a/packages/testing/containers/services/types.ts +++ b/packages/testing/containers/services/types.ts @@ -22,6 +22,7 @@ export const SERVICE_NAMES = [ 'mysql', 'localstack', 'kent', + 'postgresExporter', ] as const; export type ServiceName = (typeof SERVICE_NAMES)[number]; @@ -68,6 +69,7 @@ export interface StackConfig { env?: Record; projectName?: string; resourceQuota?: { memory?: number; cpu?: number }; + workerResourceQuota?: { memory?: number; cpu?: number }; services?: readonly ServiceName[]; /** When true, services target host machine instead of Docker-internal n8n */ external?: boolean; diff --git a/packages/testing/containers/services/victoria-metrics.ts b/packages/testing/containers/services/victoria-metrics.ts index ecf30467cab..a511cf0c9e9 100644 --- a/packages/testing/containers/services/victoria-metrics.ts +++ b/packages/testing/containers/services/victoria-metrics.ts @@ -2,6 +2,7 @@ import type { StartedNetwork } from 'testcontainers'; import { GenericContainer, Wait } from 'testcontainers'; import { TEST_CONTAINER_IMAGES } from '../test-containers'; +import { EXPORTER_PORT } from './postgres-exporter'; import type { HelperContext, Service, ServiceResult, StartContext } from './types'; const VICTORIA_METRICS_HTTP_PORT = 8428; @@ -47,12 +48,12 @@ function generateScrapeConfig(targets: ScrapeTarget[]): string { static_configs: ${targetConfigs} metrics_path: '/metrics' - scrape_interval: '5s'`); + scrape_interval: '2s'`); } return ` global: - scrape_interval: 15s + scrape_interval: 2s scrape_configs: ${scrapeConfigs.join('\n')} @@ -84,6 +85,17 @@ export const victoriaMetrics: Service = { }); } + // Add postgres-exporter scrape target when it will be started + const services = ctx.config.services ?? []; + if (ctx.usePostgres && services.includes('victoriaMetrics')) { + scrapeTargets.push({ + job: 'postgres', + instance: 'postgres', + host: 'postgres-exporter', + port: EXPORTER_PORT, + }); + } + return { scrapeTargets }; }, diff --git a/packages/testing/containers/stack.ts b/packages/testing/containers/stack.ts index e008d4b572d..a38cbe9a681 100644 --- a/packages/testing/containers/stack.ts +++ b/packages/testing/containers/stack.ts @@ -80,6 +80,7 @@ export async function createN8NStack(config: N8NConfig = {}): Promise env = {}, projectName, resourceQuota, + workerResourceQuota, services: enabledServices = [], external = false, } = config; @@ -208,6 +209,7 @@ export async function createN8NStack(config: N8NConfig = {}): Promise baseUrl: needsLoadBalancer ? undefined : baseUrl, allocatedPort: needsLoadBalancer ? undefined : allocatedMainPort, resourceQuota, + workerResourceQuota, filesToMount, }); containers.push(...n8nResult.containers); diff --git a/packages/testing/containers/test-containers.ts b/packages/testing/containers/test-containers.ts index 4c1ca30baff..5edebb93dc8 100644 --- a/packages/testing/containers/test-containers.ts +++ b/packages/testing/containers/test-containers.ts @@ -41,6 +41,7 @@ const DEFAULT_IMAGES = { kafka: 'confluentinc/cp-kafka:8.0.3', mysql: 'mysql:9.6.0', localstack: 'localstack/localstack:4.13.1', + postgresExporter: 'prometheuscommunity/postgres-exporter:v0.17.1', } as const; /** Convert camelCase to SCREAMING_SNAKE_CASE for env var names */ @@ -117,4 +118,5 @@ export const TEST_CONTAINER_IMAGES = { mysql: getImage('mysql'), ngrok: getImage('ngrok'), localstack: getImage('localstack'), + postgresExporter: getImage('postgresExporter'), } as const; diff --git a/packages/testing/playwright/package.json b/packages/testing/playwright/package.json index feabaec4290..cc7358a8c02 100644 --- a/packages/testing/playwright/package.json +++ b/packages/testing/playwright/package.json @@ -8,6 +8,7 @@ "test:e2e": "playwright test --project=*e2e*", "test:performance": "playwright test --project=performance", "test:infrastructure": "playwright test --project='*:infrastructure'", + "test:benchmark": "playwright test --project='benchmark-*'", "test:container:sqlite": "playwright test --project='sqlite:*'", "test:container:sqlite:e2e": "playwright test --project='sqlite:e2e'", "test:container:postgres": "playwright test --project='postgres:*'", @@ -37,14 +38,16 @@ "devDependencies": { "@currents/playwright": "catalog:e2e", "@n8n/api-types": "workspace:*", - "@n8n/playwright-janitor": "workspace:*", "@n8n/constants": "workspace:*", - "@n8n/workflow-sdk": "workspace:*", - "@n8n/permissions": "workspace:*", "@n8n/db": "workspace:*", + "@n8n/permissions": "workspace:*", + "@n8n/playwright-janitor": "workspace:*", + "@n8n/workflow-sdk": "workspace:*", "@playwright/cli": "catalog:e2e", "@playwright/test": "catalog:e2e", + "@types/autocannon": "^7.12.7", "@types/lodash": "catalog:", + "autocannon": "^8.0.0", "eslint-plugin-playwright": "catalog:e2e", "flatted": "catalog:", "generate-schema": "2.6.0", diff --git a/packages/testing/playwright/playwright-projects.ts b/packages/testing/playwright/playwright-projects.ts index ad21f6f3e35..394ca4c61aa 100644 --- a/packages/testing/playwright/playwright-projects.ts +++ b/packages/testing/playwright/playwright-projects.ts @@ -33,6 +33,73 @@ const CONTAINER_CONFIGS: Array<{ name: string; config: N8NConfig }> = [ }, ]; +// --- Benchmark profiles --- +// Each profile represents a real-world n8n deployment configuration. +// ONE test file runs in ALL profiles — adding a profile auto-expands coverage. + +const BENCHMARK_WORKER_COUNT = parseInt(process.env.KAFKA_LOAD_WORKERS ?? '3', 10); + +// Resource profiles matching realistic AWS instance types: +// Main: m5.large (2 vCPU, 8GB RAM) — matches staging main +// Workers: t3.medium (2 vCPU, 4GB RAM) — matches staging worker limits +export const BENCHMARK_MAIN_RESOURCES = { memory: 8, cpu: 2 }; +export const BENCHMARK_WORKER_RESOURCES = { memory: 4, cpu: 2 }; + +export const OBSERVABILITY_SERVICES = ['victoriaLogs', 'victoriaMetrics', 'vector'] as const; + +const BENCHMARK_BASE_CONFIG: N8NConfig = { + services: [...OBSERVABILITY_SERVICES], + postgres: true, + resourceQuota: BENCHMARK_MAIN_RESOURCES, + workerResourceQuota: BENCHMARK_WORKER_RESOURCES, + env: { + N8N_METRICS_INCLUDE_MESSAGE_EVENT_BUS_METRICS: 'true', + }, +}; + +const BENCHMARK_PROFILES: Array<{ name: string; config: N8NConfig }> = [ + { + name: 'direct', + config: { + ...BENCHMARK_BASE_CONFIG, + services: [...BENCHMARK_BASE_CONFIG.services!, 'kafka'], + env: { + ...BENCHMARK_BASE_CONFIG.env, + DB_POSTGRESDB_POOL_SIZE: '20', + }, + }, + }, + { + name: 'queue', + config: { + ...BENCHMARK_BASE_CONFIG, + services: [...BENCHMARK_BASE_CONFIG.services!, 'kafka'], + workers: BENCHMARK_WORKER_COUNT, + env: { + ...BENCHMARK_BASE_CONFIG.env, + N8N_METRICS_INCLUDE_QUEUE_METRICS: 'true', + }, + }, + }, + { + name: 'queue-tuned', + config: { + ...BENCHMARK_BASE_CONFIG, + services: [...BENCHMARK_BASE_CONFIG.services!, 'kafka'], + workers: BENCHMARK_WORKER_COUNT, + env: { + ...BENCHMARK_BASE_CONFIG.env, + N8N_METRICS_INCLUDE_QUEUE_METRICS: 'true', + N8N_LOG_LEVEL: 'info', + DB_POSTGRESDB_POOL_SIZE: '30', + DB_POSTGRESDB_CONNECTION_TIMEOUT: '60000', + N8N_CONCURRENCY_PRODUCTION_LIMIT: '20', + EXECUTIONS_DATA_SAVE_ON_SUCCESS: 'none', + }, + }, + }, +]; + export function getProjects(): Project[] { const isLocal = !!getBackendUrl(); const projects: Project[] = []; @@ -65,6 +132,17 @@ export function getProjects(): Project[] { }, ); } + + for (const { name, config } of BENCHMARK_PROFILES) { + projects.push({ + name: `benchmark-${name}:infrastructure`, + testDir: './tests/infrastructure/benchmarks', + workers: 1, + timeout: 600_000, + retries: 0, + use: { containerConfig: config }, + }); + } } projects.push({ diff --git a/packages/testing/playwright/playwright.config.ts b/packages/testing/playwright/playwright.config.ts index 698e992add4..d29a555cc51 100644 --- a/packages/testing/playwright/playwright.config.ts +++ b/packages/testing/playwright/playwright.config.ts @@ -116,6 +116,12 @@ export default defineConfig({ ['json', { outputFile: 'test-results.json' }], ...(process.env.CURRENTS_RECORD_KEY ? [currentsReporter(currentsConfig)] : []), ['./reporters/metrics-reporter.ts'], + ['./reporters/benchmark-summary-reporter.ts'], ] - : [['html'], ['./reporters/metrics-reporter.ts'], ['list']], + : [ + ['html'], + ['./reporters/metrics-reporter.ts'], + ['./reporters/benchmark-summary-reporter.ts'], + ['list'], + ], }); diff --git a/packages/testing/playwright/reporters/benchmark-summary-reporter.ts b/packages/testing/playwright/reporters/benchmark-summary-reporter.ts new file mode 100644 index 00000000000..1a9ebac5365 --- /dev/null +++ b/packages/testing/playwright/reporters/benchmark-summary-reporter.ts @@ -0,0 +1,205 @@ +import type { Reporter, TestCase, TestResult } from '@playwright/test/reporter'; +import { appendFileSync } from 'fs'; + +interface BenchmarkRow { + trigger: string; + suite: string; + scenario: string; + metrics: Map; +} + +interface Column { + header: string; + suffixes: string[]; + format: (value: number) => string; +} + +const COLUMNS: Column[] = [ + { + header: 'exec/s', + suffixes: ['throughput', 'exec-per-sec'], + format: (v) => v.toFixed(1), + }, + { + header: 'actions/s', + suffixes: ['actions-per-sec'], + format: (v) => v.toFixed(1), + }, + { + header: 'p50', + suffixes: ['duration-p50', 'http-latency-p50'], + format: (v) => `${v.toFixed(0)}ms`, + }, + { + header: 'p99', + suffixes: ['duration-p99', 'http-latency-p99'], + format: (v) => `${v.toFixed(0)}ms`, + }, + { + header: 'req/s', + suffixes: ['http-requests-avg'], + format: (v) => v.toFixed(1), + }, + { + header: 'errors', + suffixes: ['http-errors', 'executions-errors'], + format: (v) => String(v), + }, + { + header: 'ev lag', + suffixes: ['event-loop-lag'], + format: (v) => `${(v * 1000).toFixed(0)}ms`, + }, + { + header: 'pg tx/s', + suffixes: ['pg-tx-rate'], + format: (v) => v.toFixed(0), + }, + { + header: 'queue', + suffixes: ['queue-waiting'], + format: (v) => String(Math.round(v)), + }, +]; + +function extractTrigger(filePath: string): string { + // e.g. tests/infrastructure/benchmarks/kafka/foo.spec.ts → kafka + const match = filePath.match(/benchmarks\/([^/]+)/); + return match?.[1] ?? 'unknown'; +} + +function extractSuite(filePath: string): string { + // e.g. load.spec.ts → load, throughput.spec.ts → throughput + const filename = filePath.split('/').pop() ?? ''; + if (filename.includes('throughput')) return 'throughput'; + if (filename.includes('load')) return 'load'; + return 'other'; +} + +function extractMetricSuffix(metricName: string, scenario: string): string | null { + if (metricName.startsWith(`${scenario}-`)) { + return metricName.slice(scenario.length + 1); + } + return null; +} + +class BenchmarkSummaryReporter implements Reporter { + private rows: BenchmarkRow[] = []; + + onTestEnd(test: TestCase, result: TestResult): void { + const metricAttachments = result.attachments.filter((a) => a.name.startsWith('metric:')); + if (metricAttachments.length === 0) return; + + const scenario = test.title; + const filePath = test.location.file; + const trigger = extractTrigger(filePath); + const suite = extractSuite(filePath); + const metrics = new Map(); + + for (const attachment of metricAttachments) { + const fullName = attachment.name.replace('metric:', ''); + const suffix = extractMetricSuffix(fullName, scenario); + if (suffix) { + try { + const data = JSON.parse(attachment.body?.toString() ?? ''); + metrics.set(suffix, data.value); + } catch (error) { + console.warn(`[BenchmarkReporter] Malformed metric attachment "${fullName}":`, error); + } + } + } + + if (metrics.size > 0) { + this.rows.push({ trigger, suite, scenario, metrics }); + } + } + + onEnd(): void { + if (this.rows.length === 0) return; + + this.rows.sort( + (a, b) => + a.trigger.localeCompare(b.trigger) || + a.suite.localeCompare(b.suite) || + a.scenario.localeCompare(b.scenario), + ); + + const triggerWidth = Math.max(7, ...this.rows.map((r) => r.trigger.length)); + const suiteWidth = Math.max(5, ...this.rows.map((r) => r.suite.length)); + const scenarioWidth = Math.max(8, ...this.rows.map((r) => r.scenario.length)); + const colWidths = COLUMNS.map((col) => { + const values = this.rows.map((r) => this.resolveColumn(r, col)); + return Math.max(col.header.length, ...values.map((v) => v.length)); + }); + + const pad = (s: string, w: number) => s.padStart(w); + const padRight = (s: string, w: number) => s.padEnd(w); + + const headerParts = [ + padRight('Trigger', triggerWidth), + padRight('Suite', suiteWidth), + padRight('Scenario', scenarioWidth), + ...COLUMNS.map((col, i) => pad(col.header, colWidths[i])), + ]; + + const separator = headerParts.map((h) => '─'.repeat(h.length)); + + console.log('\n'); + console.log('Benchmark Summary'); + console.log('═'.repeat(headerParts.join(' │ ').length + 4)); + console.log(`│ ${headerParts.join(' │ ')} │`); + console.log(`├─${separator.join('─┼─')}─┤`); + + for (const row of this.rows) { + const parts = [ + padRight(row.trigger, triggerWidth), + padRight(row.suite, suiteWidth), + padRight(row.scenario, scenarioWidth), + ...COLUMNS.map((col, i) => pad(this.resolveColumn(row, col), colWidths[i])), + ]; + console.log(`│ ${parts.join(' │ ')} │`); + } + + console.log(`└─${separator.map((s) => s).join('─┴─')}─┘`); + console.log(''); + + this.writeGitHubSummary(); + } + + private writeGitHubSummary(): void { + const summaryPath = process.env.GITHUB_STEP_SUMMARY; + if (!summaryPath) return; + + const headers = ['Trigger', 'Suite', 'Scenario', ...COLUMNS.map((c) => c.header)]; + const lines: string[] = [ + '## Benchmark Summary', + '', + `| ${headers.join(' | ')} |`, + `| ${headers.map((h) => '---'.padEnd(h.length, '-')).join(' | ')} |`, + ]; + + for (const row of this.rows) { + const cells = [ + row.trigger, + row.suite, + row.scenario, + ...COLUMNS.map((col) => this.resolveColumn(row, col)), + ]; + lines.push(`| ${cells.join(' | ')} |`); + } + + lines.push(''); + appendFileSync(summaryPath, lines.join('\n')); + } + + private resolveColumn(row: BenchmarkRow, col: Column): string { + for (const suffix of col.suffixes) { + const value = row.metrics.get(suffix); + if (value !== undefined) return col.format(value); + } + return '—'; + } +} + +// eslint-disable-next-line import-x/no-default-export +export default BenchmarkSummaryReporter; diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/harness/load-harness.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/harness/load-harness.ts new file mode 100644 index 00000000000..80538aff3b2 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/harness/load-harness.ts @@ -0,0 +1,148 @@ +import { expect } from '@playwright/test'; +import type { TestInfo } from '@playwright/test'; +import type { ServiceHelpers } from 'n8n-containers/services/types'; + +import type { ApiHelpers } from '../../../../services/api-helper'; +import { + sampleExecutionDurations, + buildMetrics, + attachLoadTestResults, + waitForThroughput, + getBaselineCounter, + collectDiagnostics, + attachDiagnostics, + formatDiagnosticValue, + resolveMetricQuery, +} from '../../../../utils/benchmark'; +import type { TriggerHandle, ExecutionMetrics } from '../../../../utils/benchmark'; + +export type LoadProfile = + | { type: 'steady'; ratePerSecond: number; durationSeconds: number } + | { type: 'preloaded'; count: number }; + +export interface LoadTestOptions { + handle: TriggerHandle; + api: ApiHelpers; + services: ServiceHelpers; + testInfo: TestInfo; + load: LoadProfile; + timeoutMs: number; + /** PromQL metric to track workflow completions. Defaults to resolveMetricQuery(testInfo). */ + metricQuery?: string; +} + +/** + * Runs a single load test: creates workflow, generates load, measures completion rate and latency. + * + * Phases: create workflow → preload (if backlog) → baseline → activate → publish (if steady) → measure → report. + * + * Completion is tracked via VictoriaMetrics using the metric resolved from the project config + * (direct mode: `n8n_workflow_success_total`, queue mode: `n8n_scaling_mode_queue_jobs_completed`). + */ +export async function runLoadTest(options: LoadTestOptions): Promise { + const { handle, api, services, testInfo, load, timeoutMs } = options; + const metricQuery = options.metricQuery ?? resolveMetricQuery(testInfo); + testInfo.setTimeout(timeoutMs + 120_000); + + const obs = services.observability; + + const { workflowId, createdWorkflow } = await api.workflows.createWorkflowFromDefinition( + handle.workflow, + { makeUnique: true }, + ); + + // Phase 1: Pre-activation load (fill queue before workflow starts) + let expectedExecutions = 0; + if (load.type === 'preloaded') { + const result = await handle.preload(load.count); + console.log( + `[LOAD] Preloaded ${result.totalPublished} messages in ${result.publishDurationMs}ms`, + ); + expectedExecutions = result.totalPublished; + } + + // Phase 2: Wait for VictoriaMetrics readiness and record baseline + await obs.metrics.waitForMetric('n8n_version_info', { + timeoutMs: 30_000, + intervalMs: 2000, + predicate: (results: unknown[]) => results.length > 0, + }); + const baselineCounter = await getBaselineCounter(obs.metrics, metricQuery); + + // Phase 3: Activate workflow + // For burst tests, processing starts at activation (messages are already queued), + // so the timer must begin here to capture the full processing window. + const activationStart = Date.now(); + await api.workflows.activate(workflowId, createdWorkflow.versionId!); + await handle.waitForReady({ timeoutMs: 30_000 }); + + // Phase 4: Post-activation load (publish at controlled rate) + // For steady-state tests, n8n consumes concurrently during publishing, + // so the timer starts at publish to measure the real processing window. + let publishStart: number | undefined; + if (load.type === 'steady') { + publishStart = Date.now(); + const result = await handle.publishAtRate({ + ratePerSecond: load.ratePerSecond, + durationSeconds: load.durationSeconds, + }); + console.log( + `[LOAD] Published ${result.totalPublished} messages in ${result.actualDurationMs}ms`, + ); + expectedExecutions = result.totalPublished; + } + + // Phase 5: Wait for workflow completions via VictoriaMetrics + console.log( + `[LOAD] Waiting for ${expectedExecutions} workflow completions (timeout: ${timeoutMs}ms)`, + ); + + const throughputResult = await waitForThroughput(obs.metrics, { + expectedCount: expectedExecutions, + nodeCount: 1, + timeoutMs, + baselineValue: baselineCounter, + metricQuery, + }); + const totalDurationMs = Date.now() - (publishStart ?? activationStart); + + if (throughputResult.totalCompleted < expectedExecutions) { + console.warn( + `[LOAD] Only ${throughputResult.totalCompleted}/${expectedExecutions} completed after ${(totalDurationMs / 1000).toFixed(1)}s — results reflect partial completion`, + ); + } + + // Duration sampling is optional — may be empty when EXECUTIONS_DATA_SAVE_ON_SUCCESS=none + // hard-deletes execution records. Completion count comes from VictoriaMetrics. + const durations = await sampleExecutionDurations(api.workflows, workflowId); + const metrics = buildMetrics(throughputResult.totalCompleted, 0, totalDurationMs, durations); + + await attachLoadTestResults(testInfo, testInfo.title, metrics); + + // Diagnostics + const diagnostics = await collectDiagnostics(obs.metrics, totalDurationMs); + await attachDiagnostics(testInfo, testInfo.title, diagnostics); + const fmt = formatDiagnosticValue; + console.log( + `[DIAG] ${testInfo.title}\n` + + ` Event Loop Lag: ${fmt(diagnostics.eventLoopLag, 's')}\n` + + ` PG Transactions/s: ${fmt(diagnostics.pgTxRate, ' tx/s')}\n` + + ` PG Active Connections: ${fmt(diagnostics.pgActiveConnections)}\n` + + ` Queue Waiting: ${fmt(diagnostics.queueWaiting)}`, + ); + + console.log( + `[LOAD RESULT] ${testInfo.title}\n` + + ` Completed: ${metrics.totalCompleted}/${expectedExecutions}\n` + + ` Errors: ${metrics.totalErrors}\n` + + ` Throughput: ${metrics.throughputPerSecond.toFixed(2)} exec/s\n` + + ` Duration avg: ${metrics.avgDurationMs.toFixed(0)}ms | ` + + `p50: ${metrics.p50DurationMs.toFixed(0)}ms | ` + + `p95: ${metrics.p95DurationMs.toFixed(0)}ms | ` + + `p99: ${metrics.p99DurationMs.toFixed(0)}ms`, + ); + + expect(metrics.totalCompleted).toBeGreaterThan(0); + + return metrics; +} diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/harness/throughput-harness.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/harness/throughput-harness.ts new file mode 100644 index 00000000000..aa1fb99a4f7 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/harness/throughput-harness.ts @@ -0,0 +1,187 @@ +import { expect } from '@playwright/test'; +import type { TestInfo } from '@playwright/test'; +import type { ServiceHelpers } from 'n8n-containers/services/types'; + +import type { ApiHelpers } from '../../../../services/api-helper'; +import { + waitForThroughput, + getBaselineCounter, + attachThroughputResults, + sampleExecutionDurations, + buildMetrics, + collectDiagnostics, + attachDiagnostics, + formatDiagnosticValue, + resolveMetricQuery, +} from '../../../../utils/benchmark'; +import type { TriggerHandle, NodeOutputSize } from '../../../../utils/benchmark'; +import { attachMetric } from '../../../../utils/performance-helper'; + +export interface ThroughputTestOptions { + handle: TriggerHandle; + api: ApiHelpers; + services: ServiceHelpers; + testInfo: TestInfo; + messageCount: number; + nodeCount: number; + nodeOutputSize: NodeOutputSize; + timeoutMs: number; + pollIntervalMs?: number; + /** PromQL metric to track workflow completions. Defaults to resolveMetricQuery(testInfo). */ + metricQuery?: string; + plan?: { memory: number; cpu: number }; + workerPlan?: { memory: number; cpu: number }; +} + +function deriveProfile( + testInfo: TestInfo, + plan?: { memory: number; cpu: number }, + workerPlan?: { memory: number; cpu: number }, +) { + const name = testInfo.project.name.replace(':infrastructure', '').replace('benchmark-', ''); + const workers = + (testInfo.project.use as { containerConfig?: { workers?: number } }).containerConfig?.workers ?? + 0; + + const wp = workerPlan ?? plan; + let resourceSummary = ''; + if (plan && wp) { + resourceSummary = + workers > 0 + ? ` Mode: queue (1 main + ${workers} workers)\n` + + ` Main: ${plan.memory}GB RAM, ${plan.cpu} CPU\n` + + ` Workers: ${wp.memory}GB RAM, ${wp.cpu} CPU each\n` + + ` Total: ${(plan.memory + wp.memory * workers).toFixed(1)}GB RAM, ${plan.cpu + wp.cpu * workers} CPU` + : ` Resources: ${plan.memory}GB RAM, ${plan.cpu} CPU`; + } + + return { name, workers, resourceSummary }; +} + +/** + * Runs a single throughput test: preloads messages, activates workflow, measures drain rate. + * + * Orchestration: create workflow → preload → baseline → activate → measure throughput → diagnostics → report. + * Call this inside a `test()` body after setting up the trigger driver. + */ +export async function runThroughputTest(options: ThroughputTestOptions): Promise { + const { + handle, + api, + services, + testInfo, + messageCount, + nodeCount, + nodeOutputSize, + timeoutMs, + pollIntervalMs, + plan, + workerPlan, + } = options; + const metricQuery = options.metricQuery ?? resolveMetricQuery(testInfo); + + testInfo.setTimeout(timeoutMs + 120_000); + + const profile = deriveProfile(testInfo, plan, workerPlan); + const obs = services.observability; + + const { workflowId, createdWorkflow } = await api.workflows.createWorkflowFromDefinition( + handle.workflow, + { makeUnique: true }, + ); + + // Preload queue + const publishResult = await handle.preload(messageCount); + console.log( + `[BENCH-${profile.name}] Preloaded ${publishResult.totalPublished} messages in ${publishResult.publishDurationMs}ms`, + ); + + // Wait for VictoriaMetrics, then record baseline + await obs.metrics.waitForMetric('n8n_version_info', { + timeoutMs: 30_000, + intervalMs: 2000, + predicate: (results: unknown[]) => results.length > 0, + }); + const baselineCounter = await getBaselineCounter(obs.metrics, metricQuery); + + // Activate and wait for readiness + await api.workflows.activate(workflowId, createdWorkflow.versionId!); + await handle.waitForReady({ timeoutMs: 30_000 }); + + // Measure throughput + console.log( + `[BENCH-${profile.name}] Draining ${messageCount} messages through ${nodeCount}-node (${nodeOutputSize}) workflow (timeout: ${timeoutMs}ms)`, + ); + const result = await waitForThroughput(obs.metrics, { + expectedCount: messageCount, + nodeCount, + timeoutMs, + baselineValue: baselineCounter, + metricQuery, + pollIntervalMs, + }); + + // Attach results + await attachThroughputResults(testInfo, testInfo.title, result); + + // Execution duration sampling — provides p50/p95/p99 latency percentiles. + // May be empty when EXECUTIONS_DATA_SAVE_ON_SUCCESS=none. + const durations = await sampleExecutionDurations(api.workflows, workflowId); + if (durations.length > 0) { + const durationMetrics = buildMetrics(result.totalCompleted, 0, result.durationMs, durations); + await attachMetric( + testInfo, + `${testInfo.title}-duration-avg`, + durationMetrics.avgDurationMs, + 'ms', + ); + await attachMetric( + testInfo, + `${testInfo.title}-duration-p50`, + durationMetrics.p50DurationMs, + 'ms', + ); + await attachMetric( + testInfo, + `${testInfo.title}-duration-p95`, + durationMetrics.p95DurationMs, + 'ms', + ); + await attachMetric( + testInfo, + `${testInfo.title}-duration-p99`, + durationMetrics.p99DurationMs, + 'ms', + ); + } + + // Diagnostics + const diagnostics = await collectDiagnostics(obs.metrics, result.durationMs); + await attachDiagnostics(testInfo, testInfo.title, diagnostics); + const fmt = formatDiagnosticValue; + console.log( + `[DIAG-${profile.name}] ${testInfo.title}\n` + + ` Event Loop Lag: ${fmt(diagnostics.eventLoopLag, 's')}\n` + + ` PG Transactions/s: ${fmt(diagnostics.pgTxRate, ' tx/s')}\n` + + ` PG Rows Inserted/s: ${fmt(diagnostics.pgInsertRate, ' rows/s')}\n` + + ` PG Active Connections: ${fmt(diagnostics.pgActiveConnections)}\n` + + ` Queue Waiting: ${fmt(diagnostics.queueWaiting)}\n` + + ` Queue Active: ${fmt(diagnostics.queueActive)}\n` + + ` Queue Completed/s: ${fmt(diagnostics.queueCompletedRate, ' jobs/s')}\n` + + ` Queue Failed/s: ${fmt(diagnostics.queueFailedRate, ' jobs/s')}`, + ); + + // Summary + console.log( + `[BENCH-${profile.name} RESULT] ${testInfo.title}\n` + + ` Profile: ${profile.name}\n` + + `${profile.resourceSummary}\n` + + ` Nodes: ${nodeCount} (${nodeOutputSize}) | Messages: ${messageCount}\n` + + ` Completed: ${result.totalCompleted}/${messageCount}\n` + + ` Throughput: ${result.avgExecPerSec.toFixed(1)} exec/s | ${result.actionsPerSec.toFixed(1)} actions/s\n` + + ` Peak: ${result.peakExecPerSec.toFixed(1)} exec/s | ${result.peakActionsPerSec.toFixed(1)} actions/s\n` + + ` Duration: ${(result.durationMs / 1000).toFixed(1)}s`, + ); + + expect(result.totalCompleted).toBeGreaterThan(0); +} diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/harness/webhook-throughput-harness.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/harness/webhook-throughput-harness.ts new file mode 100644 index 00000000000..11512250d91 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/harness/webhook-throughput-harness.ts @@ -0,0 +1,174 @@ +import { expect } from '@playwright/test'; +import type { TestInfo } from '@playwright/test'; +import autocannon from 'autocannon'; +import type { ServiceHelpers } from 'n8n-containers/services/types'; + +import type { ApiHelpers } from '../../../../services/api-helper'; +import { + waitForThroughput, + getBaselineCounter, + attachThroughputResults, + collectDiagnostics, + attachDiagnostics, + formatDiagnosticValue, + resolveMetricQuery, +} from '../../../../utils/benchmark'; +import type { NodeOutputSize } from '../../../../utils/benchmark'; +import type { WebhookHandle } from '../../../../utils/benchmark/webhook-driver'; +import { attachMetric } from '../../../../utils/performance-helper'; + +export interface WebhookThroughputOptions { + handle: WebhookHandle; + api: ApiHelpers; + services: ServiceHelpers; + testInfo: TestInfo; + baseUrl: string; + nodeCount: number; + nodeOutputSize: NodeOutputSize; + connections: number; + durationSeconds: number; + timeoutMs: number; + /** PromQL metric to track workflow completions. Defaults to resolveMetricQuery(testInfo). */ + metricQuery?: string; + plan?: { memory: number; cpu: number }; + workerPlan?: { memory: number; cpu: number }; +} + +/** + * Runs a webhook throughput test using autocannon for HTTP load generation + * and VictoriaMetrics for workflow completion tracking. + * + * Phases: create workflow → activate → warm up → autocannon + VictoriaMetrics → report. + */ +export async function runWebhookThroughputTest(options: WebhookThroughputOptions): Promise { + const { + handle, + api, + services, + testInfo, + baseUrl, + nodeCount, + nodeOutputSize, + connections, + durationSeconds, + timeoutMs, + } = options; + const metricQuery = options.metricQuery ?? resolveMetricQuery(testInfo); + + testInfo.setTimeout(timeoutMs + 120_000); + + const profile = testInfo.project.name.replace(':infrastructure', '').replace('benchmark-', ''); + const obs = services.observability; + + // Phase 1: Create + activate workflow + // createWorkflowFromDefinition overwrites the webhook path and sets webhookId for proper registration + const { workflowId, createdWorkflow, webhookPath } = + await api.workflows.createWorkflowFromDefinition(handle.workflow, { + makeUnique: true, + webhookPrefix: 'bench', + }); + await api.workflows.activate(workflowId, createdWorkflow.versionId!); + + const webhookUrl = `${baseUrl}/webhook/${webhookPath}`; + + // Phase 2: Warm up — verify webhook responds (retries for async registration) + await api.webhooks.trigger(`/webhook/${webhookPath}`, { + method: 'POST', + data: handle.payload, + maxNotFoundRetries: 10, + notFoundRetryDelayMs: 500, + }); + console.log(`[WEBHOOK] Warm-up complete, webhook registered at /webhook/${webhookPath}`); + + // Phase 3: Record VictoriaMetrics baseline + await obs.metrics.waitForMetric('n8n_version_info', { + timeoutMs: 30_000, + intervalMs: 2000, + predicate: (results: unknown[]) => results.length > 0, + }); + const baselineCounter = await getBaselineCounter(obs.metrics, metricQuery); + + // Phase 4: Run autocannon + VictoriaMetrics measurement in parallel + console.log( + `[WEBHOOK] Starting ${connections} connections for ${durationSeconds}s → ${webhookUrl}\n` + + ` Workflow: ${nodeCount} nodes (${nodeOutputSize})`, + ); + + const [cannonResult, throughputResult] = await Promise.all([ + autocannon({ + url: webhookUrl, + connections, + duration: durationSeconds, + method: 'POST', + body: JSON.stringify(handle.payload), + headers: { 'Content-Type': 'application/json' }, + }), + waitForThroughput(obs.metrics, { + expectedCount: Infinity, + nodeCount, + timeoutMs: (durationSeconds + 30) * 1000, + baselineValue: baselineCounter, + metricQuery, + }), + ]); + + // Phase 5: Collect diagnostics + const diagnostics = await collectDiagnostics(obs.metrics, throughputResult.durationMs); + + // Phase 6: Attach metrics — VictoriaMetrics throughput + autocannon HTTP stats + diagnostics + await attachThroughputResults(testInfo, testInfo.title, throughputResult); + await attachMetric( + testInfo, + `${testInfo.title}-http-latency-p50`, + cannonResult.latency.p50, + 'ms', + ); + await attachMetric( + testInfo, + `${testInfo.title}-http-latency-p99`, + cannonResult.latency.p99, + 'ms', + ); + await attachMetric( + testInfo, + `${testInfo.title}-http-requests-total`, + cannonResult.requests.total, + 'count', + ); + await attachMetric( + testInfo, + `${testInfo.title}-http-requests-avg`, + cannonResult.requests.average, + 'req/s', + ); + await attachMetric( + testInfo, + `${testInfo.title}-http-errors`, + cannonResult.errors + cannonResult.non2xx, + 'count', + ); + await attachDiagnostics(testInfo, testInfo.title, diagnostics); + + // Phase 7: Log summary + const fmt = formatDiagnosticValue; + console.log( + `[DIAG-${profile}] ${testInfo.title}\n` + + ` Event Loop Lag: ${fmt(diagnostics.eventLoopLag, 's')}\n` + + ` PG Transactions/s: ${fmt(diagnostics.pgTxRate, ' tx/s')}\n` + + ` PG Active Connections: ${fmt(diagnostics.pgActiveConnections)}`, + ); + + console.log( + `[WEBHOOK-${profile} RESULT] ${testInfo.title}\n` + + ` n8n Throughput: ${throughputResult.avgExecPerSec.toFixed(1)} exec/s | ` + + `${throughputResult.actionsPerSec.toFixed(1)} actions/s\n` + + ` Peak: ${throughputResult.peakExecPerSec.toFixed(1)} exec/s | ` + + `${throughputResult.peakActionsPerSec.toFixed(1)} actions/s\n` + + ` HTTP: ${cannonResult.requests.average.toFixed(1)} req/s | ` + + `p50: ${cannonResult.latency.p50}ms | p99: ${cannonResult.latency.p99}ms\n` + + ` Errors: ${cannonResult.errors} timeouts, ${cannonResult.non2xx} non-2xx\n` + + ` Duration: ${(throughputResult.durationMs / 1000).toFixed(1)}s`, + ); + + expect(throughputResult.totalCompleted).toBeGreaterThan(0); +} diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/README.md b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/README.md new file mode 100644 index 00000000000..7cecb997dd2 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/README.md @@ -0,0 +1,131 @@ +# Kafka Benchmarks + +Two benchmark suites that measure n8n's Kafka trigger performance under different conditions. +Part of the `benchmarks/` framework — see sibling directories for other trigger types (e.g., webhook). + +## Suites + +### Load (`load.spec.ts`) + +**Question: "Can n8n keep up with incoming Kafka traffic?"** + +Measures per-execution latency (p50/p95/p99) and completion rate under realistic load patterns. Uses consumer group lag polling to track individual message consumption. + +| Scenario | What it tests | +|----------|---------------| +| `steady: 30 nodes, 10KB, 100 msg/s` | Baseline pressure — sustainable rate with realistic payloads | +| `steady: 30 nodes, 10KB, 200 msg/s` | Approaching saturation — where does latency start degrading? | +| `steady: 30 nodes, 10KB, 300 msg/s` | Saturation — should overwhelm direct mode, stress queue modes | +| `burst: 60 nodes, 1KB, drain 10k backlog` | Burst capacity — drain a backlog with no pacing | + +### Throughput (`throughput.spec.ts`) + +**Question: "What's the throughput ceiling and what degrades it?"** + +Measures sustained exec/s and actions/s via VictoriaMetrics counters. Preloads all messages before activating the workflow to measure maximum drain rate. Collects Postgres and event loop diagnostics after each run. + +ONE test file runs in ALL benchmark profiles automatically via Playwright projects. Adding a new profile or scenario auto-expands coverage. + +| Scenario | What it tests | +|----------|---------------| +| `node scaling: 10/30/60 nodes, 10KB, 10KB/node, 5k msgs` | Node count scaling curve with realistic payload (10KB in + 10KB out per node) | +| `DB pressure: 10 nodes, 1KB, 100KB/node, 5k msgs` | DB write pressure (heavy) — 100KB output per node | + +## Benchmark Profiles + +All benchmark tests (load + throughput) run in Playwright projects that represent real-world deployment configurations. Each profile provides the full container config (services, env vars, workers). One test file runs in ALL profiles automatically. + +| Profile | Mode | Workers | Log | Pool | Concurrency | Save | Matches | +|---------|------|---------|-----|------|-------------|------|---------| +| `benchmark-direct` | Direct | 0 | info | 20 | N/A | all | Self-hosted single instance | +| `benchmark-queue` | Queue | 3 | info | default | default | all | Helm chart defaults | +| `benchmark-queue-tuned` | Queue | 3 | error | 30 | 20 | none | Optimized deployment | + +Worker count is controlled via the `KAFKA_LOAD_WORKERS` env var (default: 3). + +**Adding a new profile** (e.g., multi-main): Add one entry to `BENCHMARK_PROFILES` in `playwright-projects.ts`. All tests auto-run in it. + +**Adding a new scenario**: Add a test to the relevant spec file. It auto-runs in all 3 profiles. + +**Key findings from benchmarking (10-nodes-1KB-5k noop, queue 2w):** + +| Bottleneck | Impact | Fix | +|------------|--------|-----| +| Debug logging | ~50% throughput loss | `N8N_LOG_LEVEL=error` | +| Sequential Kafka dispatch | Consumer blocked on execution completion | `parallelProcessing: true` | +| PG execution writes | Queue Completed/s doubled (45→86 jobs/s) | `EXECUTIONS_DATA_SAVE_ON_SUCCESS=none` | + +## Running + +```bash +# Build docker image first +pnpm build:docker + +# All benchmark profiles (direct + queue + queue-tuned) +pnpm --filter=n8n-playwright test:benchmark + +# Specific profile +pnpm --filter=n8n-playwright test:benchmark --project="benchmark-direct:*" +pnpm --filter=n8n-playwright test:benchmark --project="benchmark-queue:*" +pnpm --filter=n8n-playwright test:benchmark --project="benchmark-queue-tuned:*" + +# Specific scenario in specific profile +pnpm --filter=n8n-playwright test:benchmark --project="benchmark-queue:*" --grep "node scaling: 10 nodes" + +# Custom message count +BENCHMARK_MESSAGES=50000 pnpm --filter=n8n-playwright test:benchmark + +# Custom worker count (queue profiles only) +KAFKA_LOAD_WORKERS=3 pnpm --filter=n8n-playwright test:benchmark --project="benchmark-queue:*" + +# Load tests only +pnpm --filter=n8n-playwright test:benchmark --grep "Kafka Load" + +# Throughput tests only +pnpm --filter=n8n-playwright test:benchmark --grep "Kafka Throughput" +``` + +## Results + +A benchmark summary table prints at the end of every run and appears in the GitHub Actions job summary: + +``` +│ Trigger │ Suite │ Scenario │ exec/s │ actions/s │ p50 │ p95 │ p99 │ +├─────────┼────────────┼──────────────────────────┼────────┼───────────┼───────┼───────┼───────┤ +│ kafka │ load │ 10-nodes-1KB-10mps │ 90.6 │ — │ 3ms │ 5ms │ 13ms │ +│ kafka │ throughput │ 10-nodes-1KB-5k │ 142.8 │ 1427.6 │ — │ — │ — │ +``` + +- **exec/s**: Workflow executions per second +- **actions/s**: Total node executions per second (exec/s × node count) +- **p50/p95/p99**: Per-execution duration percentiles (load suite only) + +Throughput tests also log Postgres diagnostics (tx/s, rows inserted/s, active connections) and Node.js event loop lag to the console for bottleneck analysis. + +## Architecture + +The benchmark framework uses a composable architecture with four layers: + +``` +Spec files (kafka/*.spec.ts) ← wire driver + scenarios + config + ↓ passes +Generic harnesses (harness/*.ts) ← orchestrate: setup → generate → measure → report + ↓ calls +TriggerDriver interface ← encapsulates trigger-specific setup + load generation + ↓ implemented by +kafka-driver.ts ← Kafka topic/cred creation, publishing, drain tracking + ↓ uses +Shared building blocks ← workflow-builder, execution-sampler, diagnostics, throughput-measure +``` + +- **SUT** (`playwright-projects.ts`) — deployment profiles (workers, env vars, resources) +- **Generator** (`kafka-driver.ts`) — trigger-specific load production and completion tracking +- **Workflow** (`workflow-builder.ts`) — generic chain builder; any trigger node chains N nodes after it +- **Measure** (`throughput-measure.ts`, `execution-sampler.ts`, `diagnostics.ts`) — VictoriaMetrics counters, REST API latency sampling, system diagnostics + +Adding a new trigger type (e.g., webhook) requires one driver file + one spec file. +The harnesses, measurement, and reporting work unchanged. + +Both suites share the same container stack: n8n + Kafka + Postgres + postgres-exporter + VictoriaMetrics + Vector. Tests run sequentially (1 worker) to avoid resource contention. Each test creates unique topics and credentials via `nanoid()` for logical isolation. + +Direct mode tests run on a single n8n process. Queue mode tests use 1 main + N workers, controlled via `KAFKA_LOAD_WORKERS` env var (default: 3). diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-200.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-200.spec.ts new file mode 100644 index 00000000000..e3edd8c0bf4 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-200.spec.ts @@ -0,0 +1,29 @@ +import { test } from '../../../../fixtures/base'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runLoadTest } from '../harness/load-harness'; + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-load-30n-10kb-steady-200' } } }); + +test.describe( + 'Kafka Load: steady 30n/10KB/200msg', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('30 nodes, 10KB payload, steady 200 msg/s', async ({ api, services }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 30, payloadSize: '10KB' }, + }); + await runLoadTest({ + handle, + api, + services, + testInfo, + load: { type: 'steady', ratePerSecond: 200, durationSeconds: 30 }, + timeoutMs: 300_000, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-300.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-300.spec.ts new file mode 100644 index 00000000000..25b52504e36 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady-300.spec.ts @@ -0,0 +1,29 @@ +import { test } from '../../../../fixtures/base'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runLoadTest } from '../harness/load-harness'; + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-load-30n-10kb-steady-300' } } }); + +test.describe( + 'Kafka Load: steady 30n/10KB/300msg', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('30 nodes, 10KB payload, steady 300 msg/s', async ({ api, services }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 30, payloadSize: '10KB' }, + }); + await runLoadTest({ + handle, + api, + services, + testInfo, + load: { type: 'steady', ratePerSecond: 300, durationSeconds: 30 }, + timeoutMs: 300_000, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady.spec.ts new file mode 100644 index 00000000000..0dc0cc946f5 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-30n-10kb-steady.spec.ts @@ -0,0 +1,29 @@ +import { test } from '../../../../fixtures/base'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runLoadTest } from '../harness/load-harness'; + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-load-30n-10kb-steady' } } }); + +test.describe( + 'Kafka Load: steady 30n/10KB', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('30 nodes, 10KB payload, steady 100 msg/s', async ({ api, services }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 30, payloadSize: '10KB' }, + }); + await runLoadTest({ + handle, + api, + services, + testInfo, + load: { type: 'steady', ratePerSecond: 100, durationSeconds: 30 }, + timeoutMs: 300_000, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-60n-1kb-burst.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-60n-1kb-burst.spec.ts new file mode 100644 index 00000000000..774a66b572b --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/load-60n-1kb-burst.spec.ts @@ -0,0 +1,32 @@ +import { test } from '../../../../fixtures/base'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runLoadTest } from '../harness/load-harness'; + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-load-60n-1kb-burst' } } }); + +test.describe( + 'Kafka Load: burst 60n/1KB', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('60 nodes, 1KB payload, burst drain 10000 backlog', async ({ + api, + services, + }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 60, payloadSize: '1KB', partitions: 3 }, + }); + await runLoadTest({ + handle, + api, + services, + testInfo, + load: { type: 'preloaded', count: 10_000 }, + timeoutMs: 600_000, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-100kb.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-100kb.spec.ts new file mode 100644 index 00000000000..d8e49b2fd20 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-100kb.spec.ts @@ -0,0 +1,47 @@ +import { test } from '../../../../fixtures/base'; +import { + BENCHMARK_MAIN_RESOURCES, + BENCHMARK_WORKER_RESOURCES, +} from '../../../../playwright-projects'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runThroughputTest } from '../harness/throughput-harness'; + +const envMessages = parseInt(process.env.BENCHMARK_MESSAGES ?? '0', 10); + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-tp-10n-100kb' } } }); + +test.describe( + 'Kafka Throughput: 10n/100KB output', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('10 nodes, 1KB payload, 100KB output/node, 5000 msgs', async ({ + api, + services, + }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { + nodeCount: 10, + payloadSize: '1KB', + nodeOutputSize: '100KB', + partitions: 3, + }, + }); + await runThroughputTest({ + handle, + api, + services, + testInfo, + messageCount: envMessages || 5_000, + nodeCount: 10, + nodeOutputSize: '100KB', + timeoutMs: 600_000, + plan: BENCHMARK_MAIN_RESOURCES, + workerPlan: BENCHMARK_WORKER_RESOURCES, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-10kb.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-10kb.spec.ts new file mode 100644 index 00000000000..0e827483f1a --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-10n-10kb.spec.ts @@ -0,0 +1,42 @@ +import { test } from '../../../../fixtures/base'; +import { + BENCHMARK_MAIN_RESOURCES, + BENCHMARK_WORKER_RESOURCES, +} from '../../../../playwright-projects'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runThroughputTest } from '../harness/throughput-harness'; + +const envMessages = parseInt(process.env.BENCHMARK_MESSAGES ?? '0', 10); + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-tp-10n-10kb' } } }); + +test.describe( + 'Kafka Throughput: 10n/10KB', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('10 nodes, 10KB payload, 10KB output/node, 5000 msgs', async ({ + api, + services, + }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 10, payloadSize: '10KB', nodeOutputSize: '10KB', partitions: 3 }, + }); + await runThroughputTest({ + handle, + api, + services, + testInfo, + messageCount: envMessages || 5_000, + nodeCount: 10, + nodeOutputSize: '10KB', + timeoutMs: 300_000, + plan: BENCHMARK_MAIN_RESOURCES, + workerPlan: BENCHMARK_WORKER_RESOURCES, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-30n-10kb.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-30n-10kb.spec.ts new file mode 100644 index 00000000000..99fba066454 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-30n-10kb.spec.ts @@ -0,0 +1,42 @@ +import { test } from '../../../../fixtures/base'; +import { + BENCHMARK_MAIN_RESOURCES, + BENCHMARK_WORKER_RESOURCES, +} from '../../../../playwright-projects'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runThroughputTest } from '../harness/throughput-harness'; + +const envMessages = parseInt(process.env.BENCHMARK_MESSAGES ?? '0', 10); + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-tp-30n-10kb' } } }); + +test.describe( + 'Kafka Throughput: 30n/10KB', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('30 nodes, 10KB payload, 10KB output/node, 5000 msgs', async ({ + api, + services, + }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 30, payloadSize: '10KB', nodeOutputSize: '10KB', partitions: 3 }, + }); + await runThroughputTest({ + handle, + api, + services, + testInfo, + messageCount: envMessages || 5_000, + nodeCount: 30, + nodeOutputSize: '10KB', + timeoutMs: 300_000, + plan: BENCHMARK_MAIN_RESOURCES, + workerPlan: BENCHMARK_WORKER_RESOURCES, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-60n-10kb.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-60n-10kb.spec.ts new file mode 100644 index 00000000000..4cc009390c1 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/kafka/throughput-60n-10kb.spec.ts @@ -0,0 +1,42 @@ +import { test } from '../../../../fixtures/base'; +import { + BENCHMARK_MAIN_RESOURCES, + BENCHMARK_WORKER_RESOURCES, +} from '../../../../playwright-projects'; +import { kafkaDriver } from '../../../../utils/benchmark'; +import { runThroughputTest } from '../harness/throughput-harness'; + +const envMessages = parseInt(process.env.BENCHMARK_MESSAGES ?? '0', 10); + +test.use({ capability: { env: { TEST_ISOLATION: 'kafka-tp-60n-10kb' } } }); + +test.describe( + 'Kafka Throughput: 60n/10KB', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('60 nodes, 10KB payload, 10KB output/node, 5000 msgs', async ({ + api, + services, + }, testInfo) => { + const handle = await kafkaDriver.setup({ + api, + services, + scenario: { nodeCount: 60, payloadSize: '10KB', nodeOutputSize: '10KB', partitions: 3 }, + }); + await runThroughputTest({ + handle, + api, + services, + testInfo, + messageCount: envMessages || 5_000, + nodeCount: 60, + nodeOutputSize: '10KB', + timeoutMs: 600_000, + plan: BENCHMARK_MAIN_RESOURCES, + workerPlan: BENCHMARK_WORKER_RESOURCES, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-async.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-async.spec.ts new file mode 100644 index 00000000000..0171ef54196 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-async.spec.ts @@ -0,0 +1,46 @@ +import { test } from '../../../../fixtures/base'; +import { + BENCHMARK_MAIN_RESOURCES, + BENCHMARK_WORKER_RESOURCES, +} from '../../../../playwright-projects'; +import { setupWebhook } from '../../../../utils/benchmark/webhook-driver'; +import { runWebhookThroughputTest } from '../harness/webhook-throughput-harness'; + +test.use({ capability: { env: { TEST_ISOLATION: 'webhook-tp-async' } } }); + +test.describe( + 'Webhook Throughput: async', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('async: 10 nodes, 10KB payload, 10KB output/node, 50 connections, 60s', async ({ + api, + services, + backendUrl, + }, testInfo) => { + const handle = setupWebhook({ + scenario: { + nodeCount: 10, + payloadSize: '10KB', + nodeOutputSize: '10KB', + responseMode: 'onReceived', + }, + }); + await runWebhookThroughputTest({ + handle, + api, + services, + testInfo, + baseUrl: backendUrl, + nodeCount: 10, + nodeOutputSize: '10KB', + connections: 50, + durationSeconds: 60, + timeoutMs: 120_000, + plan: BENCHMARK_MAIN_RESOURCES, + workerPlan: BENCHMARK_WORKER_RESOURCES, + }); + }); + }, +); diff --git a/packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-sync.spec.ts b/packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-sync.spec.ts new file mode 100644 index 00000000000..d4c514ecaf6 --- /dev/null +++ b/packages/testing/playwright/tests/infrastructure/benchmarks/webhook/throughput-sync.spec.ts @@ -0,0 +1,46 @@ +import { test } from '../../../../fixtures/base'; +import { + BENCHMARK_MAIN_RESOURCES, + BENCHMARK_WORKER_RESOURCES, +} from '../../../../playwright-projects'; +import { setupWebhook } from '../../../../utils/benchmark/webhook-driver'; +import { runWebhookThroughputTest } from '../harness/webhook-throughput-harness'; + +test.use({ capability: { env: { TEST_ISOLATION: 'webhook-tp-sync' } } }); + +test.describe( + 'Webhook Throughput: sync', + { + annotation: [{ type: 'owner', description: 'Catalysts' }], + }, + () => { + test('sync: 10 nodes, 10KB payload, 10KB output/node, 50 connections, 60s', async ({ + api, + services, + backendUrl, + }, testInfo) => { + const handle = setupWebhook({ + scenario: { + nodeCount: 10, + payloadSize: '10KB', + nodeOutputSize: '10KB', + responseMode: 'lastNode', + }, + }); + await runWebhookThroughputTest({ + handle, + api, + services, + testInfo, + baseUrl: backendUrl, + nodeCount: 10, + nodeOutputSize: '10KB', + connections: 50, + durationSeconds: 60, + timeoutMs: 120_000, + plan: BENCHMARK_MAIN_RESOURCES, + workerPlan: BENCHMARK_WORKER_RESOURCES, + }); + }); + }, +); diff --git a/packages/testing/playwright/utils/benchmark/diagnostics.ts b/packages/testing/playwright/utils/benchmark/diagnostics.ts new file mode 100644 index 00000000000..0b323174471 --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/diagnostics.ts @@ -0,0 +1,140 @@ +import type { TestInfo } from '@playwright/test'; +import type { MetricsHelper } from 'n8n-containers'; + +import { attachMetric } from '../performance-helper'; + +export interface DiagnosticsResult { + eventLoopLag?: number; + pgTxRate?: number; + pgInsertRate?: number; + pgActiveConnections?: number; + queueWaiting?: number; + queueActive?: number; + queueCompletedRate?: number; + queueFailedRate?: number; +} + +function sumValues(results: Array<{ value: number }>): number | undefined { + if (results.length === 0) return undefined; + return results.reduce((sum, r) => sum + r.value, 0); +} + +export function formatDiagnosticValue(v: number | undefined, unit = ''): string { + return v !== undefined ? `${v.toFixed(2)}${unit}` : 'N/A'; +} + +/** Core diagnostic keys that should always be present when VictoriaMetrics has data. */ +const EXPECTED_KEYS: Array = [ + 'eventLoopLag', + 'pgTxRate', + 'pgActiveConnections', +]; + +async function queryDiagnostics( + metrics: MetricsHelper, + durationMs: number, +): Promise { + // +30s buffer accounts for VictoriaMetrics scrape interval (15s) and ingestion delay + const windowSecs = Math.ceil(durationMs / 1000) + 30; + const window = `${windowSecs}s`; + + const db = 'n8n_db'; + const [ + eventLoopLag, + pgTxRateWithTotals, + pgTxRateFallback, + pgInsertRateWithTotals, + pgInsertRateFallback, + pgActive, + queueWaiting, + queueActive, + queueCompletedRate, + queueFailedRate, + ] = await Promise.all([ + metrics.query('n8n_nodejs_eventloop_lag_seconds').catch(() => []), + metrics + .query(`rate(pg_stat_database_xact_commit_total{datname="${db}"}[${window}])`) + .catch(() => []), + metrics.query(`rate(pg_stat_database_xact_commit{datname="${db}"}[${window}])`).catch(() => []), + metrics + .query(`rate(pg_stat_database_tup_inserted_total{datname="${db}"}[${window}])`) + .catch(() => []), + metrics + .query(`rate(pg_stat_database_tup_inserted{datname="${db}"}[${window}])`) + .catch(() => []), + metrics.query(`pg_stat_activity_count{datname="${db}"}`).catch(() => []), + metrics.query('n8n_scaling_mode_queue_jobs_waiting').catch(() => []), + metrics.query('n8n_scaling_mode_queue_jobs_active').catch(() => []), + metrics.query(`rate(n8n_scaling_mode_queue_jobs_completed[${window}])`).catch(() => []), + metrics.query(`rate(n8n_scaling_mode_queue_jobs_failed[${window}])`).catch(() => []), + ]); + + const pgTxRateResult = pgTxRateWithTotals.length > 0 ? pgTxRateWithTotals : pgTxRateFallback; + const pgInsertRateResult = + pgInsertRateWithTotals.length > 0 ? pgInsertRateWithTotals : pgInsertRateFallback; + + return { + eventLoopLag: sumValues(eventLoopLag), + pgTxRate: sumValues(pgTxRateResult), + pgInsertRate: sumValues(pgInsertRateResult), + pgActiveConnections: sumValues(pgActive), + queueWaiting: sumValues(queueWaiting), + queueActive: sumValues(queueActive), + queueCompletedRate: sumValues(queueCompletedRate), + queueFailedRate: sumValues(queueFailedRate), + }; +} + +/** + * Collects system-level diagnostics from VictoriaMetrics. + * Retries when core metrics are missing — VictoriaMetrics may not have + * ingested a fresh scrape immediately after the benchmark completes. + */ +export async function collectDiagnostics( + metrics: MetricsHelper, + durationMs: number, + options: { maxRetries?: number; retryDelayMs?: number } = {}, +): Promise { + const { maxRetries = 3, retryDelayMs = 5000 } = options; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + const result = await queryDiagnostics(metrics, durationMs); + + const missing = EXPECTED_KEYS.filter((k) => result[k] === undefined); + if (missing.length === 0) return result; + + if (attempt === maxRetries) { + console.warn( + `[DIAG] Missing metrics after ${maxRetries + 1} attempts: ${missing.join(', ')}`, + ); + return result; + } + + console.log( + `[DIAG] Missing metrics (attempt ${attempt + 1}): ${missing.join(', ')} — retrying in ${retryDelayMs}ms`, + ); + await new Promise((resolve) => setTimeout(resolve, retryDelayMs)); + } + + return {}; +} + +/** + * Attaches reporter-relevant diagnostic values as test metrics. + * Only attaches values that are present (undefined = metric not available). + */ +export async function attachDiagnostics( + testInfo: TestInfo, + label: string, + diagnostics: DiagnosticsResult, +): Promise { + if (diagnostics.eventLoopLag !== undefined) { + await attachMetric(testInfo, `${label}-event-loop-lag`, diagnostics.eventLoopLag, 's'); + } + if (diagnostics.pgTxRate !== undefined) { + await attachMetric(testInfo, `${label}-pg-tx-rate`, diagnostics.pgTxRate, 'tx/s'); + } + if (diagnostics.queueWaiting !== undefined) { + await attachMetric(testInfo, `${label}-queue-waiting`, diagnostics.queueWaiting, 'count'); + } +} diff --git a/packages/testing/playwright/utils/benchmark/execution-sampler.ts b/packages/testing/playwright/utils/benchmark/execution-sampler.ts new file mode 100644 index 00000000000..83b2176b1b7 --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/execution-sampler.ts @@ -0,0 +1,101 @@ +import type { TestInfo } from '@playwright/test'; + +import type { WorkflowApiHelper } from '../../services/workflow-api-helper'; +import { attachMetric } from '../performance-helper'; +import type { ExecutionMetrics } from './types'; + +function percentile(sorted: number[], p: number): number { + if (sorted.length === 0) return 0; + const index = Math.ceil((p / 100) * sorted.length) - 1; + return sorted[Math.max(0, index)]; +} + +/** + * Fetches a sample of recent executions to calculate duration statistics. + * Retries on transient errors (e.g. 503 "Database is not ready!") since the DB + * may still be under heavy write pressure after a burst of executions. + */ +export async function sampleExecutionDurations( + workflowApi: WorkflowApiHelper, + workflowId: string, + options: { maxRetries?: number; retryDelayMs?: number } = {}, +): Promise { + const { maxRetries = 5, retryDelayMs = 3000 } = options; + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + try { + const executions = await workflowApi.getExecutions(workflowId, 100); + const durations = executions + .filter((e) => e.startedAt && e.stoppedAt) + .map((e) => new Date(e.stoppedAt!).getTime() - new Date(e.startedAt!).getTime()) + .sort((a, b) => a - b); + + if (durations.length > 0) return durations; + + // Executions not yet persisted — retry unless final attempt + if (attempt === maxRetries) { + console.warn( + `[LOAD] No execution durations found after ${maxRetries + 1} attempts — returning empty`, + ); + return []; + } + console.log( + `[LOAD] No executions found yet (attempt ${attempt + 1}), retrying in ${retryDelayMs}ms...`, + ); + } catch (error) { + if (attempt === maxRetries) { + console.warn( + `[LOAD] Failed to sample executions after ${maxRetries + 1} attempts — returning empty durations`, + ); + return []; + } + console.log( + `[LOAD] Execution sampling attempt ${attempt + 1} failed, retrying in ${retryDelayMs}ms...`, + ); + } + await new Promise((resolve) => setTimeout(resolve, retryDelayMs)); + } + + return []; +} + +export function buildMetrics( + successCount: number, + errorCount: number, + durationMs: number, + durations: number[], +): ExecutionMetrics { + const totalCompleted = successCount + errorCount; + return { + totalCompleted, + totalErrors: errorCount, + durationMs, + throughputPerSecond: durationMs > 0 ? (totalCompleted / durationMs) * 1000 : 0, + executionDurations: durations, + avgDurationMs: + durations.length > 0 ? durations.reduce((a, b) => a + b, 0) / durations.length : 0, + p50DurationMs: percentile(durations, 50), + p95DurationMs: percentile(durations, 95), + p99DurationMs: percentile(durations, 99), + }; +} + +export async function attachLoadTestResults( + testInfo: TestInfo, + label: string, + metrics: ExecutionMetrics, +): Promise { + await attachMetric(testInfo, `${label}-executions-completed`, metrics.totalCompleted, 'count'); + await attachMetric(testInfo, `${label}-executions-errors`, metrics.totalErrors, 'count'); + await attachMetric(testInfo, `${label}-throughput`, metrics.throughputPerSecond, 'exec/s'); + await attachMetric(testInfo, `${label}-total-duration`, metrics.durationMs, 'ms'); + + // Only attach duration percentiles when we have sampled data — otherwise the + // reporter would show misleading "0ms" values (e.g. when EXECUTIONS_DATA_SAVE_ON_SUCCESS=none) + if (metrics.executionDurations.length > 0) { + await attachMetric(testInfo, `${label}-duration-avg`, metrics.avgDurationMs, 'ms'); + await attachMetric(testInfo, `${label}-duration-p50`, metrics.p50DurationMs, 'ms'); + await attachMetric(testInfo, `${label}-duration-p95`, metrics.p95DurationMs, 'ms'); + await attachMetric(testInfo, `${label}-duration-p99`, metrics.p99DurationMs, 'ms'); + } +} diff --git a/packages/testing/playwright/utils/benchmark/index.ts b/packages/testing/playwright/utils/benchmark/index.ts new file mode 100644 index 00000000000..ae071f55b41 --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/index.ts @@ -0,0 +1,7 @@ +export * from './types'; +export * from './workflow-builder'; +export * from './execution-sampler'; +export * from './diagnostics'; +export * from './throughput-measure'; +export { kafkaDriver } from './kafka-driver'; +export { setupWebhook } from './webhook-driver'; diff --git a/packages/testing/playwright/utils/benchmark/kafka-driver.ts b/packages/testing/playwright/utils/benchmark/kafka-driver.ts new file mode 100644 index 00000000000..5518f3b1db0 --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/kafka-driver.ts @@ -0,0 +1,222 @@ +import { trigger } from '@n8n/workflow-sdk'; +import type { KafkaHelper } from 'n8n-containers'; +import { nanoid } from 'nanoid'; + +import type { + TriggerDriver, + TriggerHandle, + TriggerSetupContext, + PreloadResult, + PublishResult, + DrainResult, + PayloadSize, +} from './types'; +import { PAYLOAD_PROFILES, generatePayload } from './types'; +import { buildChainedWorkflow } from './workflow-builder'; + +const LAST_EXECUTION_SETTLE_MS = 3000; + +// --- Kafka-specific publishing --- + +async function publishAtRate( + kafka: KafkaHelper, + topic: string, + options: { + ratePerSecond: number; + durationSeconds: number; + payloadSize: PayloadSize; + }, +): Promise { + const { ratePerSecond, durationSeconds, payloadSize } = options; + if (ratePerSecond <= 0) throw new Error(`ratePerSecond must be > 0, got ${ratePerSecond}`); + if (durationSeconds <= 0) throw new Error(`durationSeconds must be > 0, got ${durationSeconds}`); + const payload = generatePayload(PAYLOAD_PROFILES[payloadSize]); + const intervalMs = 1000 / ratePerSecond; + const totalMessages = ratePerSecond * durationSeconds; + const startTime = Date.now(); + + for (let i = 0; i < totalMessages; i++) { + const targetTime = startTime + i * intervalMs; + const now = Date.now(); + if (now < targetTime) { + await new Promise((resolve) => setTimeout(resolve, targetTime - now)); + } + await kafka.publish(topic, { ...payload, index: i }); + } + + const actualDurationMs = Date.now() - startTime; + if (actualDurationMs > durationSeconds * 1000 * 1.1) { + console.warn( + `[LOAD] Publish rate slower than requested: took ${actualDurationMs}ms for ${durationSeconds}s target (${((actualDurationMs / (durationSeconds * 1000)) * 100).toFixed(1)}% of target)`, + ); + } + + return { totalPublished: totalMessages, actualDurationMs }; +} + +async function preloadQueue( + kafka: KafkaHelper, + topic: string, + options: { + messageCount: number; + payloadSize: PayloadSize; + }, +): Promise { + const { messageCount, payloadSize } = options; + const payload = generatePayload(PAYLOAD_PROFILES[payloadSize]); + const messages = Array.from({ length: messageCount }, (_, i) => ({ + value: { ...payload, index: i }, + })); + + // Scale batch size to stay under Kafka's message.max.bytes (default 1MB). + const payloadBytes = PAYLOAD_PROFILES[payloadSize]; + const batchSize = Math.max(1, Math.floor(900_000 / payloadBytes)); + + const startTime = Date.now(); + await kafka.publishBatch(topic, messages, { batchSize }); + + return { totalPublished: messageCount, publishDurationMs: Date.now() - startTime }; +} + +async function waitForConsumerGroupDrain( + kafka: KafkaHelper, + groupId: string, + topic: string, + options: { expectedCount: number; timeoutMs: number; pollIntervalMs?: number }, +): Promise { + const { expectedCount, timeoutMs, pollIntervalMs = 2000 } = options; + const startTime = Date.now(); + const deadline = startTime + timeoutMs; + let lastLag = -1; + + while (Date.now() < deadline) { + let lagInfo; + try { + lagInfo = await kafka.getConsumerGroupLag(groupId, topic); + } catch (error) { + console.log( + `[LOAD] Lag check error: ${error instanceof Error ? error.message : String(error)}`, + ); + await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); + continue; + } + + if (lagInfo.totalLag !== lastLag) { + const consumed = expectedCount - lagInfo.totalLag; + console.log(`[LOAD] Consumed: ${consumed}/${expectedCount} (lag=${lagInfo.totalLag})`); + lastLag = lagInfo.totalLag; + } + + if (lagInfo.totalLag === 0) { + // All messages consumed — wait briefly for last execution to finish + await new Promise((resolve) => setTimeout(resolve, LAST_EXECUTION_SETTLE_MS)); + return { drained: true, consumed: expectedCount, durationMs: Date.now() - startTime }; + } + + await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); + } + + const consumed = lastLag >= 0 ? expectedCount - lastLag : 0; + return { drained: false, consumed, durationMs: Date.now() - startTime }; +} + +// --- Kafka trigger node --- + +function createKafkaTriggerNode(options: { + topic: string; + groupId: string; + partitions: number; + credentialId: string; + credentialName: string; +}) { + return trigger({ + type: 'n8n-nodes-base.kafkaTrigger', + version: 1.1, + config: { + name: 'Kafka Trigger', + parameters: { + topic: options.topic, + groupId: options.groupId, + options: { + fromBeginning: true, + jsonParseMessage: true, + parallelProcessing: true, + sessionTimeout: 60000, + heartbeatInterval: 3000, + // Remove consumer-side bottlenecks so benchmarks measure + // n8n execution capacity, not Kafka ingestion rate. + maxInFlightRequests: 0, // 0 = unlimited (node converts to null) + partitionsConsumedConcurrently: options.partitions, + // Batch offset commits — defaults (0/undefined) commit on every + // message, adding a broker round-trip per msg that caps the main + // process consumption rate and starves queue-mode workers. + autoCommitThreshold: 50, + autoCommitInterval: 2000, + }, + }, + credentials: { + kafka: { id: options.credentialId, name: options.credentialName }, + }, + }, + }); +} + +// --- Driver implementation --- + +/** + * Kafka trigger driver for benchmarking. + * Handles topic/credential creation, message publishing, and consumer group drain tracking. + */ +export const kafkaDriver: TriggerDriver = { + requiredServices: ['kafka'], + + async setup(ctx: TriggerSetupContext): Promise { + const kafka = ctx.services.kafka; + const topic = `bench-${nanoid()}`; + const groupId = `bench-group-${nanoid()}`; + const partitions = ctx.scenario.partitions ?? 3; + const payloadSize = ctx.scenario.payloadSize; + const nodeOutputSize = ctx.scenario.nodeOutputSize ?? 'noop'; + + await kafka.createTopic(topic, partitions); + + const credential = await ctx.api.credentials.createCredential({ + name: `Kafka Bench ${nanoid()}`, + type: 'kafka', + data: { + brokers: 'kafka:9092', + clientId: `bench-${nanoid()}`, + ssl: false, + authentication: false, + }, + }); + + const kafkaTrigger = createKafkaTriggerNode({ + topic, + groupId, + partitions, + credentialId: credential.id, + credentialName: credential.name, + }); + + const label = nodeOutputSize === 'noop' ? 'noop' : `${nodeOutputSize}/node`; + const workflow = buildChainedWorkflow( + `Kafka Bench (${ctx.scenario.nodeCount} nodes, ${label})`, + kafkaTrigger, + ctx.scenario.nodeCount, + nodeOutputSize, + ); + + return { + workflow, + + preload: (count) => preloadQueue(kafka, topic, { messageCount: count, payloadSize }), + + publishAtRate: (opts) => publishAtRate(kafka, topic, { ...opts, payloadSize }), + + waitForReady: (opts) => kafka.waitForConsumerGroup(groupId, opts), + + waitForDrain: (opts) => waitForConsumerGroupDrain(kafka, groupId, topic, opts), + }; + }, +}; diff --git a/packages/testing/playwright/utils/benchmark/throughput-measure.ts b/packages/testing/playwright/utils/benchmark/throughput-measure.ts new file mode 100644 index 00000000000..481127f70de --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/throughput-measure.ts @@ -0,0 +1,227 @@ +/** + * Throughput benchmark measurement — VictoriaMetrics counter-based completion tracking. + * + * Polls a PromQL counter at regular intervals to measure sustained throughput. + * Trigger-agnostic: works with any trigger type that increments n8n_workflow_success_total. + */ +import type { TestInfo } from '@playwright/test'; +import type { MetricsHelper } from 'n8n-containers'; + +import { attachMetric } from '../performance-helper'; + +// --- Types --- + +export interface ThroughputSample { + timestamp: number; + completed: number; + delta: number; +} + +export interface ThroughputResult { + totalCompleted: number; + durationMs: number; + avgExecPerSec: number; + peakExecPerSec: number; + actionsPerSec: number; + peakActionsPerSec: number; + samples: ThroughputSample[]; +} + +// --- PromQL queries --- + +export const WORKFLOW_SUCCESS_QUERY = 'n8n_workflow_success_total'; +export const QUEUE_JOBS_COMPLETED_QUERY = 'n8n_scaling_mode_queue_jobs_completed'; + +/** + * Returns the completion metric for the current Playwright project. + * + * Currently always uses `n8n_workflow_success_total` which is emitted by both main + * and workers, aggregated across all instances by VictoriaMetrics. + * + * `n8n_scaling_mode_queue_jobs_completed` is the designed queue-mode metric but + * it depends on ScalingService.scheduleQueueMetrics() emitting `job-counts-updated` + * events at regular intervals — currently observed as 0 in CI. + */ +export function resolveMetricQuery(_testInfo: TestInfo): string { + return WORKFLOW_SUCCESS_QUERY; +} + +// --- Throughput measurement --- + +/** + * Polls VictoriaMetrics for a completion counter until it reaches the expected count. + * Records samples at each poll interval to calculate throughput. + * + * The metricQuery parameter allows switching between single-main + * (`n8n_workflow_success_total`) and queue mode (`n8n_scaling_mode_queue_jobs_completed`). + * For continuous generation tests, set expectedCount to Infinity and use timeoutMs as the run duration. + */ +export async function waitForThroughput( + metrics: MetricsHelper, + options: { + expectedCount: number; + nodeCount: number; + timeoutMs: number; + pollIntervalMs?: number; + metricQuery?: string; + baselineValue?: number; + }, +): Promise { + const { + expectedCount, + nodeCount, + timeoutMs, + pollIntervalMs = 5000, + metricQuery = WORKFLOW_SUCCESS_QUERY, + baselineValue = 0, + } = options; + + const samples: ThroughputSample[] = []; + const startTime = Date.now(); + const deadline = startTime + timeoutMs; + let lastValue = baselineValue; + let highWaterMark = baselineValue; + + while (Date.now() < deadline) { + const remaining = deadline - Date.now(); + await new Promise((resolve) => setTimeout(resolve, Math.min(pollIntervalMs, remaining))); + + let results; + try { + results = await metrics.query(`last_over_time(${metricQuery}[1m])`); + } catch (error) { + console.log( + `[THROUGHPUT] Query error: ${error instanceof Error ? error.message : String(error)}`, + ); + continue; + } + + const current = results.length > 0 ? results.reduce((sum, r) => sum + r.value, 0) : 0; + + // Monotonic guard: counters should never decrease. + // If VictoriaMetrics returns a stale/missing value, skip this sample. + if (current < highWaterMark) { + console.log( + `[THROUGHPUT] Scrape miss: counter dropped ${highWaterMark} → ${current}, skipping`, + ); + continue; + } + + highWaterMark = current; + const completed = current - baselineValue; + const delta = current - lastValue; + + samples.push({ + timestamp: Date.now(), + completed, + delta, + }); + + if (delta !== 0) { + console.log(`[THROUGHPUT] Completed: ${completed}/${expectedCount} (+${delta})`); + } + + lastValue = current; + + if (completed >= expectedCount) { + break; + } + } + + return calculateThroughput(samples, nodeCount, startTime); +} + +/** + * Reads the current value of the workflow success counter from VictoriaMetrics. + * Returns 0 if the metric hasn't been scraped yet. + */ +export async function getBaselineCounter( + metrics: MetricsHelper, + metricQuery: string = WORKFLOW_SUCCESS_QUERY, +): Promise { + try { + const results = await metrics.query(`last_over_time(${metricQuery}[1m])`); + return results.length > 0 ? results.reduce((sum, r) => sum + r.value, 0) : 0; + } catch { + return 0; + } +} + +function calculateThroughput( + samples: ThroughputSample[], + nodeCount: number, + startTime: number, +): ThroughputResult { + if (samples.length === 0) { + return { + totalCompleted: 0, + durationMs: 0, + avgExecPerSec: 0, + peakExecPerSec: 0, + actionsPerSec: 0, + peakActionsPerSec: 0, + samples: [], + }; + } + + // Duration measures actual processing time by excluding startup overhead. + // Use the last zero-progress sample as the reference start — that's the + // tightest bound on when processing actually began, regardless of whether + // completions span one poll interval or many. + const firstActiveIndex = samples.findIndex((s) => s.delta > 0); + const lastSample = samples[samples.length - 1]; + const totalCompleted = lastSample.completed; + const referenceStart = firstActiveIndex > 0 ? samples[firstActiveIndex - 1].timestamp : startTime; + const durationMs = lastSample.timestamp - referenceStart; + + // Sliding window peak: average rate over 3 consecutive intervals. + // Smooths burst noise from VictoriaMetrics scrape batching. + const PEAK_WINDOW = 3; + let peakExecPerSec = 0; + + for (let i = 0; i < samples.length; i++) { + const windowEnd = Math.min(i + PEAK_WINDOW, samples.length) - 1; + const windowStart = i; + const windowDelta = samples + .slice(windowStart, windowEnd + 1) + .reduce((sum, s) => sum + Math.max(0, s.delta), 0); + const windowStartTime = windowStart === 0 ? startTime : samples[windowStart - 1].timestamp; + const windowMs = samples[windowEnd].timestamp - windowStartTime; + if (windowMs > 0 && windowDelta > 0) { + const rate = (windowDelta / windowMs) * 1000; + peakExecPerSec = Math.max(peakExecPerSec, rate); + } + } + + const avgExecPerSec = durationMs > 0 ? (totalCompleted / durationMs) * 1000 : 0; + + return { + totalCompleted, + durationMs, + avgExecPerSec, + peakExecPerSec, + actionsPerSec: avgExecPerSec * nodeCount, + peakActionsPerSec: peakExecPerSec * nodeCount, + samples, + }; +} + +// --- Result reporting --- + +export async function attachThroughputResults( + testInfo: TestInfo, + label: string, + result: ThroughputResult, +): Promise { + await attachMetric(testInfo, `${label}-exec-per-sec`, result.avgExecPerSec, 'exec/s'); + await attachMetric(testInfo, `${label}-actions-per-sec`, result.actionsPerSec, 'actions/s'); + await attachMetric(testInfo, `${label}-peak-exec-per-sec`, result.peakExecPerSec, 'exec/s'); + await attachMetric( + testInfo, + `${label}-peak-actions-per-sec`, + result.peakActionsPerSec, + 'actions/s', + ); + await attachMetric(testInfo, `${label}-total-completed`, result.totalCompleted, 'count'); + await attachMetric(testInfo, `${label}-duration`, result.durationMs, 'ms'); +} diff --git a/packages/testing/playwright/utils/benchmark/types.ts b/packages/testing/playwright/utils/benchmark/types.ts new file mode 100644 index 00000000000..2f61135651a --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/types.ts @@ -0,0 +1,119 @@ +import type { ServiceHelpers, ServiceName } from 'n8n-containers/services/types'; +import type { IWorkflowBase } from 'n8n-workflow'; + +import type { ApiHelpers } from '../../services/api-helper'; + +// --- Payload sizes --- + +export const PAYLOAD_PROFILES = { + '1KB': 1024, + '10KB': 10240, + '100KB': 102400, +} as const; + +export type PayloadSize = keyof typeof PAYLOAD_PROFILES; + +export function generatePayload(sizeBytes: number): object { + const base = { timestamp: Date.now(), index: 0, data: '' }; + const baseSize = JSON.stringify(base).length; + const paddingSize = Math.max(0, sizeBytes - baseSize); + return { ...base, data: 'x'.repeat(paddingSize) }; +} + +// --- Node output sizes --- + +/** + * Node output modes for controlling execution data volume per node. + * - `noop`: NoOp nodes — minimal output, tests pure engine overhead. + * - `10KB` / `100KB` / `1MB`: Set nodes that add a padding field at that size. + * Tests realistic DB write pressure since execution data accumulates per node. + * Uses Set nodes (not Code nodes) to avoid task runner dependency, enabling + * clean multi-worker benchmarks. + */ +export type NodeOutputSize = 'noop' | '10KB' | '100KB' | '1MB'; + +export const OUTPUT_SIZE_BYTES: Record, number> = { + '10KB': 10_000, + '100KB': 100_000, + '1MB': 1_000_000, +}; + +// --- Execution metrics --- + +export interface ExecutionMetrics { + totalCompleted: number; + totalErrors: number; + durationMs: number; + throughputPerSecond: number; + executionDurations: number[]; + avgDurationMs: number; + p50DurationMs: number; + p95DurationMs: number; + p99DurationMs: number; +} + +// --- Trigger driver --- + +export interface PreloadResult { + totalPublished: number; + publishDurationMs: number; +} + +export interface PublishResult { + totalPublished: number; + actualDurationMs: number; +} + +export interface DrainResult { + drained: boolean; + /** Number of messages confirmed consumed (via consumer group lag tracking) */ + consumed: number; + durationMs: number; +} + +export interface TriggerSetupContext { + api: ApiHelpers; + services: ServiceHelpers; + scenario: { + nodeCount: number; + nodeOutputSize?: NodeOutputSize; + payloadSize: PayloadSize; + partitions?: number; + }; +} + +/** + * Handle returned by TriggerDriver.setup() — provides load generation + * and completion tracking for a single benchmark run. + */ +export interface TriggerHandle { + /** Workflow definition to create via API */ + workflow: Partial; + + /** Preload messages/requests before activation */ + preload(count: number): Promise; + + /** Publish at a controlled rate (steady-state load tests) */ + publishAtRate(options: { + ratePerSecond: number; + durationSeconds: number; + }): Promise; + + /** Wait for trigger to be ready after activation (e.g., consumer group joined) */ + waitForReady(options?: { timeoutMs?: number }): Promise; + + /** Wait for all messages to be consumed. Drivers without a native drain signal resolve immediately. */ + waitForDrain(options: { expectedCount: number; timeoutMs: number }): Promise; +} + +/** + * Encapsulates trigger-specific setup and load generation for benchmarking. + * Each trigger type (Kafka, Webhook, etc.) implements this once. + */ +export interface TriggerDriver { + /** Services this trigger needs in the container stack */ + readonly requiredServices: readonly ServiceName[]; + + /** Prepare trigger resources, return a handle for this benchmark run */ + setup(ctx: TriggerSetupContext): Promise; +} diff --git a/packages/testing/playwright/utils/benchmark/webhook-driver.ts b/packages/testing/playwright/utils/benchmark/webhook-driver.ts new file mode 100644 index 00000000000..a6e9d143c52 --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/webhook-driver.ts @@ -0,0 +1,66 @@ +import { trigger } from '@n8n/workflow-sdk'; +import type { IWorkflowBase } from 'n8n-workflow'; +import { nanoid } from 'nanoid'; + +import type { PayloadSize, NodeOutputSize } from './types'; +import { PAYLOAD_PROFILES, generatePayload } from './types'; +import { buildChainedWorkflow } from './workflow-builder'; + +type WebhookResponseMode = 'onReceived' | 'lastNode'; + +export interface WebhookSetupContext { + scenario: { + nodeCount: number; + payloadSize: PayloadSize; + nodeOutputSize?: NodeOutputSize; + responseMode?: WebhookResponseMode; + }; +} + +export interface WebhookHandle { + workflow: Partial; + payload: object; +} + +/** + * Sets up a webhook-triggered benchmark workflow. + * Returns a handle with the workflow definition and payload to send. + */ +export function setupWebhook(ctx: WebhookSetupContext): WebhookHandle { + const path = `bench-${nanoid()}`; + const { + nodeCount, + payloadSize, + nodeOutputSize = 'noop', + responseMode = 'onReceived', + } = ctx.scenario; + + const webhookTrigger = trigger({ + type: 'n8n-nodes-base.webhook', + version: 2, + config: { + name: 'Webhook', + parameters: { + httpMethod: 'POST', + path, + responseMode, + options: {}, + }, + }, + }); + + const label = nodeOutputSize === 'noop' ? 'noop' : `${nodeOutputSize}/node`; + const workflow = buildChainedWorkflow( + `Webhook Bench (${nodeCount} nodes, ${label}, ${responseMode})`, + webhookTrigger, + nodeCount, + nodeOutputSize, + ); + + const payload = generatePayload(PAYLOAD_PROFILES[payloadSize]); + + return { + workflow, + payload, + }; +} diff --git a/packages/testing/playwright/utils/benchmark/workflow-builder.ts b/packages/testing/playwright/utils/benchmark/workflow-builder.ts new file mode 100644 index 00000000000..1dd0c591ee1 --- /dev/null +++ b/packages/testing/playwright/utils/benchmark/workflow-builder.ts @@ -0,0 +1,64 @@ +import { workflow, node } from '@n8n/workflow-sdk'; +import type { IWorkflowBase } from 'n8n-workflow'; +import { nanoid } from 'nanoid'; + +import type { NodeOutputSize } from './types'; +import { OUTPUT_SIZE_BYTES } from './types'; + +type TriggerNode = Parameters['add']>[0]; + +export function createChainNode(index: number, outputSize: NodeOutputSize) { + if (outputSize === 'noop') { + return node({ + type: 'n8n-nodes-base.noOp', + version: 1, + config: { name: `NoOp ${index}` }, + }); + } + + return node({ + type: 'n8n-nodes-base.set', + version: 3.4, + config: { + name: `Set ${index}`, + parameters: { + assignments: { + assignments: [ + { + id: 'payload', + name: 'payload', + value: `={{ 'x'.repeat(${OUTPUT_SIZE_BYTES[outputSize]}) }}`, + type: 'string', + }, + ], + }, + includeOtherFields: true, + options: {}, + }, + }, + }); +} + +/** + * Builds a workflow: triggerNode → N chained nodes. + * The trigger can be any node type — Kafka, Webhook, Cron, etc. + */ +export function buildChainedWorkflow( + name: string, + triggerNode: TriggerNode, + nodeCount: number, + nodeOutputSize: NodeOutputSize = 'noop', +): Partial { + if (nodeCount <= 0) throw new Error(`nodeCount must be > 0, got ${nodeCount}`); + + const [first, ...rest] = Array.from({ length: nodeCount }, (_, i) => + createChainNode(i + 1, nodeOutputSize), + ); + + const wf = workflow(nanoid(), name); + wf.add( + rest.reduce((chain, n) => chain.to(n), (triggerNode as ReturnType).to(first)), + ); + + return wf.toJSON() as Partial; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5413b644c7f..71973283fa6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -574,7 +574,7 @@ importers: version: 3.0.1 axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) jest-mock-extended: specifier: ^3.0.4 version: 3.0.4(jest@29.7.0(@types/node@20.19.21)(ts-node@10.9.2(@swc/core@1.15.8(@swc/helpers@0.5.17))(@types/node@20.19.21)(typescript@5.9.2)))(typescript@5.9.2) @@ -797,7 +797,7 @@ importers: version: 4.0.7 axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) dotenv: specifier: 17.2.3 version: 17.2.3 @@ -832,7 +832,7 @@ importers: dependencies: axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) devDependencies: '@n8n/typescript-config': specifier: workspace:* @@ -1748,7 +1748,7 @@ importers: version: link:../eslint-plugin-community-nodes axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) eslint: specifier: 'catalog:' version: 9.29.0(jiti@2.6.1) @@ -2032,7 +2032,7 @@ importers: version: 1.11.0 axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) bcryptjs: specifier: 2.4.3 version: 2.4.3 @@ -2390,7 +2390,7 @@ importers: version: 10.36.0 axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) callsites: specifier: 'catalog:' version: 3.1.0 @@ -2863,7 +2863,7 @@ importers: version: link:../../../@n8n/utils axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) flatted: specifier: 'catalog:' version: 3.2.7 @@ -3184,7 +3184,7 @@ importers: version: 1.1.4 axios: specifier: 1.13.5 - version: 1.13.5 + version: 1.13.5(debug@4.4.3) bowser: specifier: 2.11.0 version: 2.11.0 @@ -3891,9 +3891,15 @@ importers: '@playwright/test': specifier: catalog:e2e version: 1.58.0 + '@types/autocannon': + specifier: ^7.12.7 + version: 7.12.7 '@types/lodash': specifier: 'catalog:' version: 4.17.17 + autocannon: + specifier: ^8.0.0 + version: 8.0.0 eslint-plugin-playwright: specifier: catalog:e2e version: 2.2.2(eslint@9.29.0(jiti@2.6.1)) @@ -4107,6 +4113,9 @@ packages: '@apm-js-collab/tracing-hooks@0.3.1': resolution: {integrity: sha512-Vu1CbmPURlN5fTboVuKMoJjbO5qcq9fA5YXpskx3dXe/zTBvjODFoerw+69rVBlRLrJpwPqSDqEuJDEKIrTldw==} + '@assemblyscript/loader@0.19.23': + resolution: {integrity: sha512-ulkCYfFbYj01ie1MDOyxv2F6SpRN1TOj7fQxbP07D6HmeR+gr2JLSmINKjga2emB+b1L2KGrFKBTc+e00p54nw==} + '@authenio/xml-encryption@2.0.2': resolution: {integrity: sha512-cTlrKttbrRHEw3W+0/I609A2Matj5JQaRvfLtEIGZvlN0RaPi+3ANsMeqAyCAVlH/lUIW2tmtBlSMni74lcXeg==} engines: {node: '>=12'} @@ -6870,6 +6879,9 @@ packages: '@microsoft/tsdoc@0.15.1': resolution: {integrity: sha512-4aErSrCR/On/e5G2hDP0wjooqDdauzEbIq8hIkIe5pXV0rtWJZvdCEKL0ykZxex+IxIwBp0eGeV48hQN07dXtw==} + '@minimistjs/subarg@1.0.0': + resolution: {integrity: sha512-Q/ONBiM2zNeYUy0mVSO44mWWKYM3UHuEK43PKIOzJCbvUnPoMH1K+gk3cf1kgnCVJFlWmddahQQCmrmBGlk9jQ==} + '@miragejs/pretender-node-polyfill@0.1.2': resolution: {integrity: sha512-M/BexG/p05C5lFfMunxo/QcgIJnMT2vDVCd00wNqK2ImZONIlEETZwWJu1QtLxtmYlSHlCFl3JNzp0tLe7OJ5g==} @@ -9163,6 +9175,9 @@ packages: '@types/asn1@0.2.0': resolution: {integrity: sha512-5TMxIpYbIA9c1J0hYQjQDX3wr+rTgQEAXaW2BI8ECM8FO53wSW4HFZplTalrKSHuZUc76NtXcePRhwuOHqGD5g==} + '@types/autocannon@7.12.7': + resolution: {integrity: sha512-Pd4nPf7wRpacULa6D/EC9x3CwzFQXwA0z5WFuik/fvJjW44V3WzBTM3jtt8nSBoflUNgswPiMCtgrr1bwnAcMg==} + '@types/aws4@1.11.2': resolution: {integrity: sha512-x0f96eBPrCCJzJxdPbUvDFRva4yPpINJzTuXXpmS2j9qLUpF2nyGzvXPlRziuGbCsPukwY4JfuO+8xwsoZLzGw==} @@ -10509,6 +10524,10 @@ packages: resolution: {integrity: sha512-kNOjDqAh7px0XWNI+4QbzoiR/nTkHAWNud2uvnJquD1/x5a7EQZMJT0AczqK0Qn67oY/TTQ1LbUKajZpp3I9tQ==} engines: {node: '>=8.0.0'} + autocannon@8.0.0: + resolution: {integrity: sha512-fMMcWc2JPFcUaqHeR6+PbmEpTxCrPZyBUM95oG4w3ngJ8NfBNas/ZXA+pTHXLqJ0UlFVTcy05GC25WxKx/M20A==} + hasBin: true + autoprefixer@10.4.19: resolution: {integrity: sha512-BaENR2+zBZ8xXhM4pUaKUxlVdxZ0EZhjvbopwnXmxRUfqDmwSpC2lAi/QXvx7NRdPCo1WKEcEF6mV64si1z4Ew==} engines: {node: ^10 || ^12 || >=14} @@ -10965,6 +10984,9 @@ packages: resolution: {integrity: sha512-kWWXztvZ5SBQV+eRgKFeh8q5sLuZY2+8WUIzlxWVTg+oGwY14qylx1KbKzHd8P6ZYkAg0xyIDU9JMHhyJMZ1jw==} engines: {node: '>=10'} + char-spinner@1.0.1: + resolution: {integrity: sha512-acv43vqJ0+N0rD+Uw3pDHSxP30FHrywu2NO6/wBaHChJIizpDeBUd6NjqhNhy9LGaEAhZAXn46QzmlAvIWd16g==} + character-parser@2.2.0: resolution: {integrity: sha512-+UqJQjFEFaTAs3bNsF2j2kEN1baG/zghZbdqoYEDxGZtJo9LBzl1A+m0D4n3qKx8N2FNv8/Xp6yV9mQmBuptaw==} @@ -11449,6 +11471,9 @@ packages: resolution: {integrity: sha512-fkdfq+b+AHI4cKdhZlppHveI/mgz2qpiYxcm+t5E5TsxX7QrLS1VE0+7GENEk9z0EeGPcpSciGv6ez24duWhwQ==} engines: {node: '>=18.x'} + cross-argv@2.0.0: + resolution: {integrity: sha512-YIaY9TR5Nxeb8SMdtrU8asWVM4jqJDNDYlKV21LxtYcfNJhp1kEsgSa6qXwXgzN0WQWGODps0+TlGp2xQSHwOg==} + cross-env@7.0.3: resolution: {integrity: sha512-+/HKd6EgcQCJGh2PSjZuUitQBQynKor4wrFbRg4DtAgS1aWO+gU52xpH7M9ScGgXSYmAVS9bIJ8EzuaGw0oNAw==} engines: {node: '>=10.14', npm: '>=6', yarn: '>=1'} @@ -13136,6 +13161,9 @@ packages: engines: {node: '>=0.4.7'} hasBin: true + has-async-hooks@1.0.0: + resolution: {integrity: sha512-YF0VPGjkxr7AyyQQNykX8zK4PvtEDsUJAPqwu06UFz1lb6EvI53sPh5H1kWxg8NXI5LsfRCZ8uX9NkYDZBb/mw==} + has-bigints@1.1.0: resolution: {integrity: sha512-R3pbpkcIqv2Pm3dUwgjclDRVmWpTJW2DcMzcIhEXEx1oh/CEMObMm3KLmRJOdvhM7o4uQBnwr8pzRK2sJWIqfg==} engines: {node: '>= 0.4'} @@ -13191,6 +13219,13 @@ packages: resolution: {integrity: sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ==} engines: {node: '>= 0.4'} + hdr-histogram-js@3.0.1: + resolution: {integrity: sha512-l3GSdZL1Jr1C0kyb461tUjEdrRPZr8Qry7jByltf5JGrA0xvqOSrxRBfcrJqqV/AMEtqqhHhC6w8HW0gn76tRQ==} + engines: {node: '>=14'} + + hdr-histogram-percentiles-obj@3.0.0: + resolution: {integrity: sha512-7kIufnBqdsBGcSZLPJwqHT3yhk1QTsSlFsVD3kx5ixH/AlgBs9yM1q6DPhXZ8f8gtdqgh7N7/5btRLpQsS2gHw==} + he@1.2.0: resolution: {integrity: sha512-F/1DnUGPopORZi0ni+CvrCgHQ5FyEAHRLSApuYWMmrbSwoN2Mn/7k+Gl38gJnR7yyDZk6WLXwiGod1JOWNDKGw==} hasBin: true @@ -13289,6 +13324,9 @@ packages: resolution: {integrity: sha512-4FbRdAX+bSdmo4AUFuS0WNiPz8NgFt+r8ThgNWmlrjQjt1Q7ZR9+zTlce2859x4KSXrwIsaeTqDoKQmtP8pLmQ==} engines: {node: '>= 0.8'} + http-parser-js@0.5.10: + resolution: {integrity: sha512-Pysuw9XpUq5dVc/2SMHpuTY01RFl8fttgcyunjL7eEMhGM3cI4eOmiCycJDVCo/7O7ClfQD3SaI6ftDzqOXYMA==} + http-proxy-agent@4.0.1: resolution: {integrity: sha512-k0zdNgqWTGA6aeIRVpvfVob4fL52dTfaehylg0Y4UvSySvOq/Y+BOyPrgpUrA7HylqvU8vIZGsRuXmspskV0Tg==} engines: {node: '>= 6'} @@ -13341,6 +13379,9 @@ packages: humanize-ms@1.2.1: resolution: {integrity: sha512-Fl70vYtsAFb/C06PTS9dZBo7ihau+Tu/DNCk/OyHhea07S+aeMWpFFkUaXRa8fI+ScZbEI8dfSxwY7gxZ9SAVQ==} + hyperid@3.3.0: + resolution: {integrity: sha512-7qhCVT4MJIoEsNcbhglhdmBKb09QtcmJNiIQGq7js/Khf5FtQQ9bzcAuloeqBeee7XD7JqDeve9KNlQya5tSGQ==} + ibm-cloud-sdk-core@5.3.2: resolution: {integrity: sha512-YhtS+7hGNO61h/4jNShHxbbuJ1TnDqiFKQzfEaqePnonOvv8NnxWxOk92FlKKCCzZNOT34Gnd7WCLVJTntwEFQ==} engines: {node: '>=18'} @@ -14640,6 +14681,9 @@ packages: lodash.camelcase@4.3.0: resolution: {integrity: sha512-TwuEnCnxbc3rAvhf/LbG7tJUDzhqXyFnv3dtzLOPgCG/hODL7WFnsbwktkD7yUV0RrreP/l1PALq/YSg6VvjlA==} + lodash.chunk@4.2.0: + resolution: {integrity: sha512-ZzydJKfUHJwHa+hF5X66zLFCBrWn5GeF28OHEr4WVWtNDXlQ/IjWKPBiikqKo2ne0+v6JgCgJ0GzJp8k8bHC7w==} + lodash.clonedeep@4.5.0: resolution: {integrity: sha512-H5ZhCF25riFd9uB5UCkVKo61m3S/xZk1x4wA6yp/L3RFP6Z/eHH1ymQcGLo7J3GMPfm0V/7m1tryHuGVxpqEBQ==} @@ -14649,6 +14693,9 @@ packages: lodash.defaults@4.2.0: resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==} + lodash.flatten@4.4.0: + resolution: {integrity: sha512-C5N2Z3DgnnKr0LOpv/hKCgKdb7ZZwafIrsesve6lmzvZIRZRGaZ/l6Q8+2W7NaT+ZwO3fFlSCzCzrDCFdJfZ4g==} + lodash.flattendeep@4.4.0: resolution: {integrity: sha512-uHaJFihxmJcEX3kT4I23ABqKKalJ/zDrDg0lsFtc1h+3uw49SIJ5beyhx5ExVRti3AvKoOJngIj7xz3oylPdWQ==} @@ -14834,6 +14881,9 @@ packages: engines: {node: '>=12.0.0'} hasBin: true + manage-path@2.0.0: + resolution: {integrity: sha512-NJhyB+PJYTpxhxZJ3lecIGgh4kwIY2RAh44XvAz9UlqthlQwtPBf62uBVR8XaD8CRuSjQ6TnZH2lNJkbLPZM2A==} + map-stream@0.1.0: resolution: {integrity: sha512-CkYQrPYZfWnu/DAmVCpTSX/xHpKZ80eKh2lAkyA6AJTef6bW+6JpbQZN5rofum7da+SyN1bi5ctTm+lTfcCW3g==} @@ -15741,6 +15791,10 @@ packages: resolution: {integrity: sha512-737ZY3yNnXy37FHkQxPzt4UZ2UWPWiCZWLvFZ4fu5cueciegX0zGPnrlY6bwRg4FdQOe9YU8MkmJwGhoMybl8A==} engines: {node: '>= 0.8'} + on-net-listen@1.1.2: + resolution: {integrity: sha512-y1HRYy8s/RlcBvDUwKXSmkODMdx4KSuIvloCnQYJ2LdBBC1asY4HtfhXwe3UWknLakATZDnbzht2Ijw3M1EqFg==} + engines: {node: '>=9.4.0 || ^8.9.4'} + once@1.4.0: resolution: {integrity: sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==} @@ -16941,6 +16995,9 @@ packages: resolution: {integrity: sha512-l+sSefzHpj5qimhFSE5a8nufZYAM3sBSVMAPtYkmC+4EH2anSGaEMXSD0izRQbu9nfyQ9y5JrVmp7E8oZrUjvA==} engines: {node: '>=8'} + retimer@3.0.0: + resolution: {integrity: sha512-WKE0j11Pa0ZJI5YIk0nflGI7SQsfl2ljihVy7ogh7DeQSeYAUi0ubZ/yEueGtDfUPk6GH5LRw1hBdLq4IwUBWA==} + retry-axios@2.6.0: resolution: {integrity: sha512-pOLi+Gdll3JekwuFjXO3fTq+L9lzMQGcSq7M5gIjExcl3Gu1hd4XXuf5o3+LuSBsaULQH7DiNbsqPd1chVpQGQ==} engines: {node: '>=10.7.0'} @@ -17933,6 +17990,10 @@ packages: resolution: {integrity: sha512-9phl76Cqm6FhSX9Xe1ZUAMLtm1BLkKj2Qd5ApyWkXzsMRaA7dgr81kf4wJmQf/hAvg8EEyJxDo3du/0KlhPiKQ==} engines: {node: '>=0.6.0'} + timestring@6.0.0: + resolution: {integrity: sha512-wMctrWD2HZZLuIlchlkE2dfXJh7J2KDI9Dwl+2abPYg0mswQHfOAyQW3jJg1pY5VfttSINZuKcXoB3FGypVklA==} + engines: {node: '>=8'} + tiny-invariant@1.3.3: resolution: {integrity: sha512-+FbBPE1o9QAYvviau/qC5SE3caw21q3xkvWKBtja5vgqOWIHHJ3ioaq1VPfn/Szqctz2bU/oYeKd9/z5BL+PVg==} @@ -18598,6 +18659,9 @@ packages: uuencode@0.0.4: resolution: {integrity: sha512-yEEhCuCi5wRV7Z5ZVf9iV2gWMvUZqKJhAs1ecFdKJ0qzbyaVelmsE3QjYAamehfp9FKLiZbKldd+jklG3O0LfA==} + uuid-parse@1.1.0: + resolution: {integrity: sha512-OdmXxA8rDsQ7YpNVbKSJkNzTw2I+S5WsbMDnCtIWSQaosNAcWtFuI/YK1TjzUI6nbkgiqEyh8gWngfcv8Asd9A==} + uuid@10.0.0: resolution: {integrity: sha512-8XkAphELsDnEGrDxUOHB3RGvXz6TeuYSGEZBOjtTtPm2lwhGBjLgOzLHB63IUWfBpNucQjND6d3AOudO+H3RWQ==} hasBin: true @@ -19570,6 +19634,8 @@ snapshots: transitivePeerDependencies: - supports-color + '@assemblyscript/loader@0.19.23': {} + '@authenio/xml-encryption@2.0.2': dependencies: '@xmldom/xmldom': 0.8.10 @@ -22336,7 +22402,7 @@ snapshots: '@codspeed/core@4.0.1': dependencies: - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) find-up: 6.3.0 form-data: 4.0.4 node-gyp-build: 4.8.4 @@ -23997,7 +24063,7 @@ snapshots: '@azure/core-auth': 1.10.1 '@azure/msal-node': 3.8.4 '@microsoft/agents-activity': 1.2.3 - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) jsonwebtoken: 9.0.3 jwks-rsa: 3.2.2 object-path: 0.11.8 @@ -24041,6 +24107,10 @@ snapshots: '@microsoft/tsdoc@0.15.1': {} + '@minimistjs/subarg@1.0.0': + dependencies: + minimist: 1.2.8 + '@miragejs/pretender-node-polyfill@0.1.2': {} '@mistralai/mistralai@1.10.0': @@ -25348,7 +25418,7 @@ snapshots: '@rudderstack/rudder-sdk-node@3.0.0': dependencies: - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) axios-retry: 4.5.0(axios@1.13.5) component-type: 2.0.0 join-component: 1.1.0 @@ -26743,6 +26813,10 @@ snapshots: dependencies: '@types/node': 20.19.21 + '@types/autocannon@7.12.7': + dependencies: + '@types/node': 20.19.21 + '@types/aws4@1.11.2': dependencies: '@types/node': 20.17.57 @@ -28415,6 +28489,32 @@ snapshots: atomic-sleep@1.0.0: {} + autocannon@8.0.0: + dependencies: + '@minimistjs/subarg': 1.0.0 + chalk: 4.1.2 + char-spinner: 1.0.1 + cli-table3: 0.6.5 + color-support: 1.1.3 + cross-argv: 2.0.0 + form-data: 4.0.4 + has-async-hooks: 1.0.0 + hdr-histogram-js: 3.0.1 + hdr-histogram-percentiles-obj: 3.0.0 + http-parser-js: 0.5.10 + hyperid: 3.3.0 + lodash.chunk: 4.2.0 + lodash.clonedeep: 4.5.0 + lodash.flatten: 4.4.0 + manage-path: 2.0.0 + on-net-listen: 1.1.2 + pretty-bytes: 5.6.0 + progress: 2.0.3 + reinterval: 1.1.0 + retimer: 3.0.0 + semver: 7.7.3 + timestring: 6.0.0 + autoprefixer@10.4.19(postcss@8.4.49): dependencies: browserslist: 4.24.4 @@ -28439,17 +28539,9 @@ snapshots: axios-retry@4.5.0(axios@1.13.5): dependencies: - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) is-retry-allowed: 2.2.0 - axios@1.13.5: - dependencies: - follow-redirects: 1.15.11(debug@4.4.1) - form-data: 4.0.4 - proxy-from-env: 1.1.0 - transitivePeerDependencies: - - debug - axios@1.13.5(debug@4.4.3): dependencies: follow-redirects: 1.15.11(debug@4.4.3) @@ -29015,6 +29107,8 @@ snapshots: char-regex@1.0.2: {} + char-spinner@1.0.1: {} + character-parser@2.2.0: dependencies: is-regex: 1.2.1 @@ -29249,8 +29343,7 @@ snapshots: color-name: 1.1.4 simple-swizzle: 0.2.2 - color-support@1.1.3: - optional: true + color-support@1.1.3: {} color@3.2.1: dependencies: @@ -29517,6 +29610,8 @@ snapshots: '@types/luxon': 3.7.1 luxon: 3.7.2 + cross-argv@2.0.0: {} + cross-env@7.0.3: dependencies: cross-spawn: 7.0.6 @@ -31669,6 +31764,8 @@ snapshots: optionalDependencies: uglify-js: 3.17.4 + has-async-hooks@1.0.0: {} + has-bigints@1.1.0: {} has-flag@3.0.0: {} @@ -31719,6 +31816,14 @@ snapshots: dependencies: function-bind: 1.1.2 + hdr-histogram-js@3.0.1: + dependencies: + '@assemblyscript/loader': 0.19.23 + base64-js: 1.5.1 + pako: 1.0.11 + + hdr-histogram-percentiles-obj@3.0.0: {} + he@1.2.0: {} header-case@2.0.4: @@ -31840,6 +31945,8 @@ snapshots: statuses: 2.0.2 toidentifier: 1.0.1 + http-parser-js@0.5.10: {} + http-proxy-agent@4.0.1: dependencies: '@tootallnate/once': 1.1.2 @@ -31918,6 +32025,12 @@ snapshots: dependencies: ms: 2.1.3 + hyperid@3.3.0: + dependencies: + buffer: 5.7.1 + uuid: 8.3.2 + uuid-parse: 1.1.0 + ibm-cloud-sdk-core@5.3.2: dependencies: '@types/debug': 4.1.12 @@ -32018,7 +32131,7 @@ snapshots: infisical-node@1.3.0: dependencies: - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) dotenv: 16.6.1 tweetnacl: 1.0.3 tweetnacl-util: 0.15.1 @@ -33722,12 +33835,16 @@ snapshots: lodash.camelcase@4.3.0: {} + lodash.chunk@4.2.0: {} + lodash.clonedeep@4.5.0: {} lodash.debounce@4.0.8: {} lodash.defaults@4.2.0: {} + lodash.flatten@4.4.0: {} + lodash.flattendeep@4.4.0: {} lodash.get@4.4.2: {} @@ -33939,6 +34056,8 @@ snapshots: underscore: 1.13.8 xmlbuilder: 10.1.1 + manage-path@2.0.0: {} + map-stream@0.1.0: {} mappersmith@2.45.0: {} @@ -35101,6 +35220,8 @@ snapshots: on-headers@1.1.0: {} + on-net-listen@1.1.2: {} + once@1.4.0: dependencies: wrappy: 1.0.2 @@ -35724,7 +35845,7 @@ snapshots: posthog-node@3.2.1: dependencies: - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) rusha: 0.8.14 transitivePeerDependencies: - debug @@ -36430,9 +36551,11 @@ snapshots: onetime: 5.1.2 signal-exit: 3.0.7 + retimer@3.0.0: {} + retry-axios@2.6.0(axios@1.13.5): dependencies: - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) retry-request@7.0.2(encoding@0.1.13): dependencies: @@ -37050,7 +37173,7 @@ snapshots: asn1.js: 5.4.1 asn1.js-rfc2560: 5.0.1(asn1.js@5.4.1) asn1.js-rfc5280: 3.0.0 - axios: 1.13.5 + axios: 1.13.5(debug@4.4.3) big-integer: 1.6.52 bignumber.js: 9.1.2 binascii: 0.0.2 @@ -37844,6 +37967,8 @@ snapshots: dependencies: setimmediate: 1.0.5 + timestring@6.0.0: {} + tiny-invariant@1.3.3: {} tinybench@2.9.0: {} @@ -38531,6 +38656,8 @@ snapshots: uuencode@0.0.4: {} + uuid-parse@1.1.0: {} + uuid@10.0.0: {} uuid@11.1.0: {}