mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 21:49:36 +02:00
chore: demo dashboard and multi-worker support for the seats PoC harness (branch-only)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Michael Siega
parent
e391d08317
commit
c467a2e0cf
77
packages/testing/poc-trigger-service/demo-dashboard.sh
Executable file
77
packages/testing/poc-trigger-service/demo-dashboard.sh
Executable file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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": {},
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user