From c467a2e0cf1dee61ba8e4f61eabf8bee73c1095d Mon Sep 17 00:00:00 2001 From: Michael Siega Date: Mon, 31 Aug 2026 16:00:03 +0200 Subject: [PATCH] chore: demo dashboard and multi-worker support for the seats PoC harness (branch-only) Co-Authored-By: Claude Fable 5 --- .../poc-trigger-service/demo-dashboard.sh | 77 +++++++++++++++++++ .../testing/poc-trigger-service/run-worker.sh | 18 +++-- packages/testing/poc-trigger-service/seed.sh | 16 ++-- 3 files changed, 99 insertions(+), 12 deletions(-) create mode 100755 packages/testing/poc-trigger-service/demo-dashboard.sh diff --git a/packages/testing/poc-trigger-service/demo-dashboard.sh b/packages/testing/poc-trigger-service/demo-dashboard.sh new file mode 100755 index 00000000000..c5c85cb051f --- /dev/null +++ b/packages/testing/poc-trigger-service/demo-dashboard.sh @@ -0,0 +1,77 @@ +#!/usr/bin/env bash +# Live pipeline dashboard for recording the demo: +# Kafka -> [mains: seat holders consuming] -> queue -> [workers: executing] +# Counts come from instance logs (consumption/execution markers) and Postgres. +# ./demo-dashboard.sh [refresh-seconds] +set -euo pipefail +POC_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +INTERVAL="${1:-1}" + +BOLD=$'\e[1m'; DIM=$'\e[2m'; RESET=$'\e[0m' +BLUE=$'\e[34m'; GREEN=$'\e[32m'; YELLOW=$'\e[33m'; MAGENTA=$'\e[35m' + +psql_q() { + docker exec -i n8n-poc-postgres-1 psql -U n8n -d n8n -tA -c "$1" 2>/dev/null || echo '?' +} + +bar() { # count max-width + local count="$1" width=40 total="$2" fill + [ "$total" -eq 0 ] && total=1 + fill=$(( count * width / total )) + [ "$fill" -gt "$width" ] && fill=$width + printf '%s%s' "$(printf '█%.0s' $(seq 1 $((fill + 1))))" "$(printf "${DIM}·%.0s${RESET}" $(seq 1 $((width - fill))))" +} + +while true; do + TOTAL=$(psql_q 'SELECT count(*) FROM execution_entity') + LAST10=$(psql_q "SELECT count(*) FROM execution_entity WHERE \"createdAt\" > now() - interval '10 seconds'") + + SEATS=$(psql_q "SELECT w.name || '|' || s.\"seatIndex\" || '|' || coalesce(s.\"holderId\",'-') || '|' || s.\"leaseEpoch\" || '|' || coalesce(s.\"actualState\",'-') FROM workflow_trigger_seat s JOIN workflow_entity w ON w.id = s.\"workflowId\" WHERE s.\"desiredState\"='active' ORDER BY w.name, s.\"seatIndex\"") + + OUT="" + OUT+="${BOLD} KAFKA ─▶ MAINS (trigger seats) ─▶ QUEUE ─▶ WORKERS${RESET}\n" + OUT+=" $(date '+%H:%M:%S') total executions: ${BOLD}${TOTAL}${RESET} last 10s: ${BOLD}${LAST10}${RESET}\n\n" + + OUT+="${BOLD}${BLUE} MAINS — Kafka messages consumed (log: 'Received trigger')${RESET}\n" + MAINS_TOTAL=0 + declare -a MAIN_LINES=() + for f in "$POC_DIR"/.poc-data/main-*.log; do + [ -f "$f" ] || continue + name=$(basename "$f" .log) + count=$(grep -c 'Received trigger for workflow' "$f" 2>/dev/null || true); count=${count:-0} + MAINS_TOTAL=$((MAINS_TOTAL + count)) + MAIN_LINES+=("$name $count") + done + for line in "${MAIN_LINES[@]:-}"; do + [ -z "$line" ] && continue + name=${line% *}; count=${line##* } + OUT+="$(printf ' %-9s %6d %s' "$name" "$count" "$(bar "$count" "$MAINS_TOTAL")")\n" + done + + OUT+="\n${BOLD}${GREEN} WORKERS — executions run (log: 'Worker finished execution')${RESET}\n" + WORKERS_TOTAL=0 + declare -a WORKER_LINES=() + for f in "$POC_DIR"/.poc-data/worker-*.log; do + [ -f "$f" ] || continue + name=$(basename "$f" .log) + count=$(grep -c 'Worker finished execution' "$f" 2>/dev/null || true); count=${count:-0} + WORKERS_TOTAL=$((WORKERS_TOTAL + count)) + WORKER_LINES+=("$name $count") + done + for line in "${WORKER_LINES[@]:-}"; do + [ -z "$line" ] && continue + name=${line% *}; count=${line##* } + OUT+="$(printf ' %-9s %6d %s' "$name" "$count" "$(bar "$count" "$WORKERS_TOTAL")")\n" + done + + OUT+="\n${BOLD}${MAGENTA} TRIGGER SEATS${RESET}\n" + OUT+="$(printf ' %-18s %-4s %-24s %-6s %s' 'workflow' 'seat' 'holder' 'epoch' 'state')\n" + while IFS='|' read -r wf seat holder epoch state; do + [ -z "$wf" ] && continue + OUT+="$(printf ' %-18s %-4s %-24s %-6s %s' "${wf:0:18}" "$seat" "$holder" "$epoch" "$state")\n" + done <<< "$SEATS" + + clear + printf '%b\n' "$OUT" + sleep "$INTERVAL" +done diff --git a/packages/testing/poc-trigger-service/run-worker.sh b/packages/testing/poc-trigger-service/run-worker.sh index f2cf23c9ff6..b16b1f891ff 100755 --- a/packages/testing/poc-trigger-service/run-worker.sh +++ b/packages/testing/poc-trigger-service/run-worker.sh @@ -1,13 +1,19 @@ #!/usr/bin/env bash -# Run one local queue-mode worker (executions need at least one). +# Run local queue-mode worker #N (executions need at least one). +# ./run-worker.sh -> worker-1 +# ./run-worker.sh 2 -> worker-2 set -euo pipefail +N="${1:-1}" source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/lib.sh" -export N8N_USER_FOLDER="$HOME/.n8n-poc-seats/worker-1" -mkdir -p "$N8N_USER_FOLDER" -# Worker health-check server; must not collide with the mains (5678-5680). -export QUEUE_HEALTH_CHECK_PORT=5681 +export N8N_USER_FOLDER="$HOME/.n8n-poc-seats/worker-$N" +mkdir -p "$N8N_USER_FOLDER" "$POC_DIR/.poc-data" +# Worker health-check server; mains use 5678-5680, so workers go to 569x. +export QUEUE_HEALTH_CHECK_PORT=$((5690 + N)) +# The task broker defaults to 5679 (collides with main-2 and other workers). +export N8N_RUNNERS_BROKER_PORT=$((5700 + N)) -echo "[poc] worker-1" +echo "[poc] worker-$N" cd "$REPO_ROOT" +echo $$ > "$POC_DIR/.poc-data/worker-$N.pid" exec node packages/cli/bin/n8n worker diff --git a/packages/testing/poc-trigger-service/seed.sh b/packages/testing/poc-trigger-service/seed.sh index 587e1b90b43..32b5de77270 100755 --- a/packages/testing/poc-trigger-service/seed.sh +++ b/packages/testing/poc-trigger-service/seed.sh @@ -9,6 +9,9 @@ POC_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" SEATS="${1:-2}" PARTITIONS="${2:-4}" +TOPIC="${TOPIC:-poc-events}" +GROUP="${GROUP:-poc-seats}" +WF_NAME="${WF_NAME:-PoC Kafka Seats}" BASE="${N8N_BASE:-http://localhost:5678}" EMAIL="poc@n8n.io" PASSWORD="PocPassword1" @@ -25,10 +28,10 @@ api() { # method path [json-body] fi } -echo "[seed] ensuring kafka topic poc-events ($PARTITIONS partitions)" +echo "[seed] ensuring kafka topic $TOPIC ($PARTITIONS partitions)" docker exec n8n-poc-kafka-1 /opt/kafka/bin/kafka-topics.sh \ --bootstrap-server localhost:9092 --create --if-not-exists \ - --topic poc-events --partitions "$PARTITIONS" --replication-factor 1 >/dev/null + --topic "$TOPIC" --partitions "$PARTITIONS" --replication-factor 1 >/dev/null echo "[seed] owner setup / login" SETUP=$(api POST /owner/setup "{\"email\":\"$EMAIL\",\"firstName\":\"Poc\",\"lastName\":\"Owner\",\"password\":\"$PASSWORD\"}" || true) @@ -42,11 +45,12 @@ CRED_ID=$(echo "$CRED" | python3 -c 'import json,sys; print(json.load(sys.stdin) echo "[seed] credential id: $CRED_ID" echo "[seed] demo workflow (seatCount=$SEATS)" -WF_BODY=$(python3 - "$CRED_ID" "$SEATS" <<'EOF' +WF_BODY=$(python3 - "$CRED_ID" "$SEATS" "$TOPIC" "$GROUP" "$WF_NAME" <<'EOF' import json, sys cred_id, seats = sys.argv[1], int(sys.argv[2]) +topic, group, wf_name = sys.argv[3], sys.argv[4], sys.argv[5] wf = { - "name": "PoC Kafka Seats", + "name": wf_name, "nodes": [ { "id": "kafka-trigger-node", @@ -55,8 +59,8 @@ wf = { "typeVersion": 2, "position": [0, 0], "parameters": { - "topic": "poc-events", - "groupId": "poc-seats", + "topic": topic, + "groupId": group, "resolveOffset": "onCompletion", "seatCount": seats, "options": {},