189 lines
9.2 KiB
Bash
Executable File
189 lines
9.2 KiB
Bash
Executable File
#!/bin/sh
|
|
# 在一次性 MySQL 中复现集中触发后的持久任务积压曲线。
|
|
# 样本故意以有界截止结束；退出 0 表示“预期的非零工具结果已核对”，不是任务全部完成。
|
|
set -eu
|
|
|
|
TASK_DOCKER=${DOCKER:-docker}
|
|
TASK_GO=${GO:-go}
|
|
TASK_MYSQL_IMAGE=${ORBIT_TEST_MYSQL_IMAGE:-mysql:8.4}
|
|
TASK_RUN_TIMEOUT=${ORBIT_BACKLOG_RUN_TIMEOUT:-180s}
|
|
TASK_NAME="orbit-schedule-backlog-$$"
|
|
TASK_DIR=$(mktemp -d "${TMPDIR:-/tmp}/orbit-schedule-backlog.XXXXXX")
|
|
TASK_SERVER_PID=
|
|
TASK_LOAD_PID=
|
|
|
|
cleanup() {
|
|
if [ -n "$TASK_LOAD_PID" ]; then
|
|
kill "$TASK_LOAD_PID" >/dev/null 2>&1 || true
|
|
wait "$TASK_LOAD_PID" 2>/dev/null || true
|
|
fi
|
|
if [ -n "$TASK_SERVER_PID" ]; then
|
|
kill "$TASK_SERVER_PID" >/dev/null 2>&1 || true
|
|
wait "$TASK_SERVER_PID" 2>/dev/null || true
|
|
fi
|
|
"$TASK_DOCKER" container stop "$TASK_NAME" >/dev/null 2>&1 || true
|
|
"$TASK_DOCKER" container rm "$TASK_NAME" >/dev/null 2>&1 || true
|
|
if [ "${ORBIT_TEST_KEEP_ARTIFACTS:-0}" = 1 ]; then
|
|
TASK_KEEP_DIR=${ORBIT_TEST_KEEP_ARTIFACTS_PATH:-/tmp/orbit-schedule-backlog-last}
|
|
mkdir -p "$TASK_KEEP_DIR"
|
|
for task_artifact in "$TASK_DIR"/*; do
|
|
if [ -f "$task_artifact" ]; then cp "$task_artifact" "$TASK_KEEP_DIR/"; fi
|
|
done
|
|
fi
|
|
for task_artifact in "$TASK_DIR"/*; do
|
|
if [ -f "$task_artifact" ]; then unlink "$task_artifact"; fi
|
|
done
|
|
rmdir "$TASK_DIR" 2>/dev/null || true
|
|
}
|
|
trap cleanup EXIT HUP INT TERM
|
|
|
|
umask 077
|
|
TASK_DB_PASSWORD=$(openssl rand -hex 24)
|
|
"$TASK_DOCKER" run -d --name "$TASK_NAME" \
|
|
-e MYSQL_ROOT_PASSWORD="$TASK_DB_PASSWORD" \
|
|
-e MYSQL_DATABASE=orbit_test \
|
|
-p 127.0.0.1::3306 "$TASK_MYSQL_IMAGE" >/dev/null
|
|
TASK_READY=0
|
|
for task_attempt in $(seq 1 60); do
|
|
if "$TASK_DOCKER" exec -e MYSQL_PWD="$TASK_DB_PASSWORD" "$TASK_NAME" \
|
|
mysql -uroot --protocol=tcp -e 'SELECT 1' >/dev/null 2>&1; then
|
|
TASK_READY=1
|
|
break
|
|
fi
|
|
sleep 1
|
|
done
|
|
if [ "$TASK_READY" -ne 1 ]; then
|
|
printf '%s\n' 'Disposable MySQL failed to become ready.' >&2
|
|
"$TASK_DOCKER" logs --tail 30 "$TASK_NAME" >&2 || true
|
|
exit 1
|
|
fi
|
|
TASK_DB_PORT=$("$TASK_DOCKER" port "$TASK_NAME" 3306/tcp | sed 's/.*://')
|
|
TASK_DSN="root:$TASK_DB_PASSWORD@tcp(127.0.0.1:$TASK_DB_PORT)/orbit_test"
|
|
|
|
"$TASK_GO" build -o "$TASK_DIR/orbit" ./cmd/orbit
|
|
"$TASK_GO" build -o "$TASK_DIR/orbit-migrate" ./cmd/orbit-migrate
|
|
"$TASK_GO" build -o "$TASK_DIR/orbit-admin" ./cmd/orbit-admin
|
|
(cd sdk/go && GOWORK=off "$TASK_GO" build -o "$TASK_DIR/orbit-schedule-load" ./cmd/orbit-schedule-load)
|
|
ORBIT_MYSQL_DSN="$TASK_DSN" "$TASK_DIR/orbit-migrate" up >/dev/null
|
|
ORBIT_MYSQL_DSN="$TASK_DSN" "$TASK_DIR/orbit-admin" bootstrap \
|
|
--application schedule-backlog --namespace test --environment test --queue schedule \
|
|
--worker-access --schedule-access --output "$TASK_DIR/credentials.json" >/dev/null
|
|
|
|
ORBIT_MYSQL_DSN="$TASK_DSN" ORBIT_HTTP_ADDR=127.0.0.1:0 ORBIT_METRICS_ADDR=127.0.0.1:0 "$TASK_DIR/orbit" >"$TASK_DIR/server.log" 2>&1 &
|
|
TASK_SERVER_PID=$!
|
|
TASK_HTTP_PORT=
|
|
TASK_METRICS_PORT=
|
|
for task_attempt in $(seq 1 30); do
|
|
TASK_HTTP_PORT=$(sed -n 's/.*"address":"127.0.0.1:\([0-9]*\)".*/\1/p' "$TASK_DIR/server.log" | head -n 1)
|
|
TASK_METRICS_PORT=$(sed -n 's/.*"metrics_address":"127.0.0.1:\([0-9]*\)".*/\1/p' "$TASK_DIR/server.log" | head -n 1)
|
|
if [ -n "$TASK_HTTP_PORT" ] && [ -n "$TASK_METRICS_PORT" ]; then break; fi
|
|
if ! kill -0 "$TASK_SERVER_PID" 2>/dev/null; then cat "$TASK_DIR/server.log" >&2; exit 1; fi
|
|
sleep 1
|
|
done
|
|
if [ -z "$TASK_HTTP_PORT" ] || [ -z "$TASK_METRICS_PORT" ]; then
|
|
printf '%s\n' 'Orbit HTTP or metrics server failed to start.' >&2
|
|
exit 1
|
|
fi
|
|
|
|
query() {
|
|
"$TASK_DOCKER" exec -e MYSQL_PWD="$TASK_DB_PASSWORD" "$TASK_NAME" \
|
|
mysql -uroot --protocol=tcp --batch --skip-column-names orbit_test -e "$1"
|
|
}
|
|
: >"$TASK_DIR/curve.jsonl"
|
|
: >"$TASK_DIR/metrics.jsonl"
|
|
export ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_DIR/credentials.json"
|
|
set +e
|
|
"$TASK_DIR/orbit-schedule-load" -schedules 32 -requests 8 -workers 1 -capacity 1 \
|
|
-start-delay 5s -run-timeout "$TASK_RUN_TIMEOUT" -poll-interval 100ms \
|
|
>"$TASK_DIR/report.json" 2>"$TASK_DIR/tool.stderr" &
|
|
TASK_LOAD_PID=$!
|
|
set -e
|
|
while kill -0 "$TASK_LOAD_PID" 2>/dev/null; do
|
|
TASK_AT=$(date -u +%Y-%m-%dT%H:%M:%S.000Z)
|
|
TASK_ROW=$(query 'SELECT (SELECT COUNT(*) FROM pulse_tasks WHERE state="WAITING"),(SELECT COUNT(*) FROM pulse_tasks WHERE state="LEASED"),(SELECT COUNT(*) FROM pulse_tasks WHERE state="RUNNING"),(SELECT COUNT(*) FROM pulse_tasks WHERE state="SUCCEEDED"),(SELECT COUNT(*) FROM pulse_schedule_occurrences),(SELECT COUNT(*) FROM pulse_schedules WHERE paused=FALSE)' 2>/dev/null || true)
|
|
set -- $TASK_ROW
|
|
if [ "$#" -eq 6 ]; then
|
|
printf '{"at":"%s","waiting":%s,"leased":%s,"running":%s,"succeeded":%s,"occurrences":%s,"unpaused_schedules":%s}\n' "$TASK_AT" "$1" "$2" "$3" "$4" "$5" "$6" >>"$TASK_DIR/curve.jsonl"
|
|
fi
|
|
if curl -fsS --max-time 3 "http://127.0.0.1:$TASK_METRICS_PORT/metrics" >"$TASK_DIR/metrics.current" 2>/dev/null; then
|
|
TASK_PROCESS_CPU=$(ps -p "$TASK_SERVER_PID" -o %cpu= | awk '{print $1}')
|
|
TASK_PROCESS_RSS=$(ps -p "$TASK_SERVER_PID" -o rss= | awk '{print $1}')
|
|
TASK_DB_STATS=$("$TASK_DOCKER" stats --no-stream --format '{{.CPUPerc}}|{{.MemUsage}}|{{.MemPerc}}' "$TASK_NAME" 2>/dev/null || true)
|
|
METRICS_AT="$TASK_AT" PROCESS_CPU="$TASK_PROCESS_CPU" PROCESS_RSS_KB="$TASK_PROCESS_RSS" DB_CONTAINER_STATS="$TASK_DB_STATS" python3 - "$TASK_DIR/metrics.current" >>"$TASK_DIR/metrics.jsonl" <<'PY'
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
with open(sys.argv[1], encoding="utf-8") as f:
|
|
lines = f.read().splitlines()
|
|
values = {}
|
|
for line in lines:
|
|
if line.startswith("#") or " " not in line:
|
|
continue
|
|
name, value = line.rsplit(" ", 1)
|
|
try:
|
|
values[name] = float(value)
|
|
except ValueError:
|
|
pass
|
|
stats = os.environ.get("DB_CONTAINER_STATS", "")
|
|
cpu, memory, memory_pct = (stats.split("|", 2) + ["", "", ""])[:3]
|
|
row = {
|
|
"at": os.environ["METRICS_AT"],
|
|
"server_cpu_percent": float(os.environ.get("PROCESS_CPU") or 0),
|
|
"server_rss_kb": int(float(os.environ.get("PROCESS_RSS_KB") or 0)),
|
|
"mysql_cpu_percent": cpu,
|
|
"mysql_memory": memory,
|
|
"mysql_memory_percent": memory_pct,
|
|
"db_connections_open": values.get("orbit_db_connections_open", 0),
|
|
"db_connections_in_use": values.get("orbit_db_connections_in_use", 0),
|
|
"db_connections_idle": values.get("orbit_db_connections_idle", 0),
|
|
"db_wait_count_total": values.get("orbit_db_wait_count_total", 0),
|
|
"db_wait_duration_seconds_total": values.get("orbit_db_wait_duration_seconds_total", 0),
|
|
"waiting_tasks": values.get('orbit_pulse_tasks_state{state="WAITING"}', 0),
|
|
"running_tasks": values.get('orbit_pulse_tasks_state{state="RUNNING"}', 0),
|
|
"business_snapshot_duration_count": values.get("orbit_business_snapshot_duration_seconds_count", 0),
|
|
"business_snapshot_duration_sum": values.get("orbit_business_snapshot_duration_seconds_sum", 0),
|
|
}
|
|
print(json.dumps(row, separators=(",", ":")))
|
|
PY
|
|
fi
|
|
sleep 2
|
|
done
|
|
set +e
|
|
wait "$TASK_LOAD_PID"
|
|
TASK_LOAD_CODE=$?
|
|
set -e
|
|
TASK_LOAD_PID=
|
|
if [ "$TASK_LOAD_CODE" -ne 1 ]; then
|
|
printf 'expected orbit-schedule-load exit 1, got %s\n' "$TASK_LOAD_CODE" >&2
|
|
cat "$TASK_DIR/tool.stderr" >&2 || true
|
|
exit 1
|
|
fi
|
|
TASK_FINAL=$(query 'SELECT JSON_OBJECT("foreign_key_count",(SELECT COUNT(*) FROM information_schema.referential_constraints WHERE constraint_schema=DATABASE()),"task_count",(SELECT COUNT(*) FROM pulse_tasks),"succeeded_count",(SELECT COUNT(*) FROM pulse_tasks WHERE state="SUCCEEDED"),"occurrence_count",(SELECT COUNT(*) FROM pulse_schedule_occurrences),"unpaused_schedules",(SELECT COUNT(*) FROM pulse_schedules WHERE paused=FALSE));')
|
|
TASK_VERIFY=$(ORBIT_MYSQL_DSN="$TASK_DSN" "$TASK_DIR/orbit-admin" verify)
|
|
python3 - "$TASK_DIR/report.json" "$TASK_FINAL" <<'PY'
|
|
import json
|
|
import sys
|
|
|
|
with open(sys.argv[1], encoding="utf-8") as f:
|
|
report = json.load(f)
|
|
database = json.loads(sys.argv[2])
|
|
if report["expected_occurrences"] != 256 or report["occurrences"] != 256 or report["fired"] != 256:
|
|
raise SystemExit("not all bounded schedule occurrences were persisted")
|
|
if report["complete"] or report["errors"].get("RUN_TIMEOUT") != 1 or report["unverified"] <= 0:
|
|
raise SystemExit("deadline negative sample did not preserve an incomplete denominator")
|
|
if not all(plan["pause_confirmed"] for plan in report["plans"]):
|
|
raise SystemExit("not every created schedule was confirmed paused")
|
|
if database["task_count"] != 256 or database["occurrence_count"] != 256 or database["unpaused_schedules"] != 0 or database["foreign_key_count"] != 0:
|
|
raise SystemExit("final database evidence is inconsistent")
|
|
PY
|
|
TASK_POINTS=$(wc -l <"$TASK_DIR/curve.jsonl" | tr -d ' ')
|
|
if [ "${ORBIT_TEST_KEEP_ARTIFACTS:-0}" = 1 ]; then
|
|
mkdir -p "${ORBIT_TEST_KEEP_ARTIFACTS_PATH:-/tmp/orbit-schedule-backlog-last}"
|
|
cp "$TASK_DIR/report.json" "${ORBIT_TEST_KEEP_ARTIFACTS_PATH:-/tmp/orbit-schedule-backlog-last}/sample-256-deadline.json"
|
|
cp "$TASK_DIR/curve.jsonl" "${ORBIT_TEST_KEEP_ARTIFACTS_PATH:-/tmp/orbit-schedule-backlog-last}/curve.jsonl"
|
|
cp "$TASK_DIR/metrics.jsonl" "${ORBIT_TEST_KEEP_ARTIFACTS_PATH:-/tmp/orbit-schedule-backlog-last}/metrics.jsonl"
|
|
printf '%s\n' "$TASK_FINAL" >"${ORBIT_TEST_KEEP_ARTIFACTS_PATH:-/tmp/orbit-schedule-backlog-last}/database.json"
|
|
fi
|
|
printf '{"tool_exit":%s,"verify":%s,"database":%s,"curve_points":%s}\n' "$TASK_LOAD_CODE" "$TASK_VERIFY" "$TASK_FINAL" "$TASK_POINTS"
|