ad/packmate
1
0
Fork 0
mirror of https://github.com/Av0idD4rk/packmate.git synced 2026-10-23 14:31:01 +00:00
Code Issues Packages Projects Releases Wiki Activity
69294f77ab
packmate/stress/network/run.sh

651 lines
27 KiB
Bash
Raw Blame History

#!/bin/sh
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
REPO_DIR=$(CDPATH= cd -- "$SCRIPT_DIR/../.." && pwd)
COMPOSE_FILE="$SCRIPT_DIR/docker-compose.yml"
PROJECT_RAW=${PACKMATE_NETWORK_STRESS_PROJECT:-"packmate-network-stress-${CI_JOB_ID:-local}"}
PROJECT=$(printf '%s' "$PROJECT_RAW" | tr -cd 'a-zA-Z0-9_-' | tr 'A-Z' 'a-z')
RUN_LABEL_RAW=${PACKMATE_NETWORK_STRESS_RUN_ID:-"$(date -u +%Y%m%dT%H%M%SZ)-$$"}
RUN_LABEL=$(printf '%s' "$RUN_LABEL_RAW" | tr -cd 'a-zA-Z0-9_.-')
if [ -z "$PROJECT" ] || [ -z "$RUN_LABEL" ]; then
echo "stress project and run ID must contain letters, digits, dot, underscore or dash" >&2
exit 2
fi
REPORT_DIR="$REPO_DIR/build/reports/network-stress/$RUN_LABEL"
REPORT_DIR_DOCKER="$REPORT_DIR"
if [ -n "${MSYSTEM:-}" ]; then
COMPOSE_FILE=$(cygpath -w "$COMPOSE_FILE")
REPORT_DIR_DOCKER=$(cygpath -w "$REPORT_DIR")
fi
BASELINE_STREAMS=${PACKMATE_NETWORK_STRESS_STREAMS:-250000}
LIVE_STREAMS=${PACKMATE_NETWORK_STRESS_LIVE_STREAMS:-50000}
SHARDS=${PACKMATE_NETWORK_STRESS_SHARDS:-16}
CONCURRENCY=${PACKMATE_NETWORK_STRESS_CONCURRENCY:-128}
FRAMES=${PACKMATE_NETWORK_STRESS_FRAMES_PER_STREAM:-8}
PAYLOAD_BYTES=${PACKMATE_NETWORK_STRESS_PAYLOAD_BYTES:-1024}
DRAIN_TIMEOUT=${PACKMATE_NETWORK_STRESS_DRAIN_TIMEOUT_SECONDS:-1800}
CLEANUP_TIMEOUT=${PACKMATE_NETWORK_STRESS_CLEANUP_TIMEOUT_SECONDS:-1800}
MAX_API_P99_MS=${PACKMATE_NETWORK_STRESS_MAX_API_P99_MS:-500}
MAX_CAPTURE_QUEUE=${PACKMATE_NETWORK_STRESS_MAX_CAPTURE_QUEUE:-500000}
CAPTURE_WORKERS=${PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS:-8}
CAPTURE_QUEUE_CAPACITY=${PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY:-250000}
CAPTURE_BUFFER_SIZE_BYTES=${PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES:-67108864}
TCP_TIMEOUT_SECONDS=${PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS:-40}
UDP_TIMEOUT_SECONDS=${PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS:-20}
MIN_STREAMS_PER_SECOND=${PACKMATE_NETWORK_STRESS_MIN_STREAMS_PER_SECOND:-500}
MIN_CLEANUP_STREAMS_PER_SECOND=${PACKMATE_NETWORK_STRESS_MIN_CLEANUP_STREAMS_PER_SECOND:-$MIN_STREAMS_PER_SECOND}
MAX_INGEST_DRAIN_SECONDS=${PACKMATE_NETWORK_STRESS_MAX_INGEST_DRAIN_SECONDS:-120}
MIN_LIVE_DURATION_SECONDS=${PACKMATE_NETWORK_STRESS_MIN_LIVE_DURATION_SECONDS:-20}
FAVORITE_STREAMS=${PACKMATE_NETWORK_STRESS_FAVORITE_STREAMS:-100}
STORAGE_MULTIPLIER=${PACKMATE_NETWORK_STRESS_STORAGE_MULTIPLIER:-3}
TARGET_IP=${PACKMATE_NETWORK_STRESS_TARGET_IP:-172.31.250.10}
PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS=${PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS:-10}
PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES=${PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES:-1440}
PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS=$CAPTURE_WORKERS
PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY=$CAPTURE_QUEUE_CAPACITY
PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES=$CAPTURE_BUFFER_SIZE_BYTES
PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS=$TCP_TIMEOUT_SECONDS
PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS=$UDP_TIMEOUT_SECONDS
export PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES
export PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY
export PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES
export PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS
PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED=false
export PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED
require_positive() {
case "$1" in ''|*[!0-9]*|0) echo "$2 must be a positive integer" >&2; exit 2 ;; esac
}
for setting in \
"$BASELINE_STREAMS:PACKMATE_NETWORK_STRESS_STREAMS" \
"$LIVE_STREAMS:PACKMATE_NETWORK_STRESS_LIVE_STREAMS" \
"$SHARDS:PACKMATE_NETWORK_STRESS_SHARDS" \
"$CONCURRENCY:PACKMATE_NETWORK_STRESS_CONCURRENCY" \
"$FRAMES:PACKMATE_NETWORK_STRESS_FRAMES_PER_STREAM" \
"$PAYLOAD_BYTES:PACKMATE_NETWORK_STRESS_PAYLOAD_BYTES" \
"$MAX_CAPTURE_QUEUE:PACKMATE_NETWORK_STRESS_MAX_CAPTURE_QUEUE" \
"$CAPTURE_WORKERS:PACKMATE_NETWORK_STRESS_CAPTURE_WORKERS" \
"$CAPTURE_QUEUE_CAPACITY:PACKMATE_NETWORK_STRESS_CAPTURE_QUEUE_CAPACITY" \
"$CAPTURE_BUFFER_SIZE_BYTES:PACKMATE_NETWORK_STRESS_CAPTURE_BUFFER_SIZE_BYTES" \
"$TCP_TIMEOUT_SECONDS:PACKMATE_NETWORK_STRESS_TCP_TIMEOUT_SECONDS" \
"$UDP_TIMEOUT_SECONDS:PACKMATE_NETWORK_STRESS_UDP_TIMEOUT_SECONDS" \
"$MIN_STREAMS_PER_SECOND:PACKMATE_NETWORK_STRESS_MIN_STREAMS_PER_SECOND" \
"$MIN_CLEANUP_STREAMS_PER_SECOND:PACKMATE_NETWORK_STRESS_MIN_CLEANUP_STREAMS_PER_SECOND" \
"$MAX_INGEST_DRAIN_SECONDS:PACKMATE_NETWORK_STRESS_MAX_INGEST_DRAIN_SECONDS" \
"$MIN_LIVE_DURATION_SECONDS:PACKMATE_NETWORK_STRESS_MIN_LIVE_DURATION_SECONDS" \
"$FAVORITE_STREAMS:PACKMATE_NETWORK_STRESS_FAVORITE_STREAMS" \
"$STORAGE_MULTIPLIER:PACKMATE_NETWORK_STRESS_STORAGE_MULTIPLIER" \
"$PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS:PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS" \
"$PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES:PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES"
do
require_positive "${setting%%:*}" "${setting#*:}"
done
if [ "$PAYLOAD_BYTES" -lt 32 ]; then
echo "PACKMATE_NETWORK_STRESS_PAYLOAD_BYTES must be at least 32" >&2
exit 2
fi
if [ "$FAVORITE_STREAMS" -gt "$BASELINE_STREAMS" ]; then
echo "favorite stream count cannot exceed baseline streams" >&2
exit 2
fi
mkdir -p "$REPORT_DIR"
compose() {
MSYS_NO_PATHCONV=1 docker compose --profile tools -p "$PROJECT" -f "$COMPOSE_FILE" "$@"
}
db_scalar() {
output=$(compose exec -T db psql -v ON_ERROR_STOP=1 -U packmate -d packmate -Atqc "$1") || return $?
printf '%s' "$output" | tr -d '\r'
}
db_integer() {
value=$(db_scalar "$1") || return $?
case "$value" in
''|*[!0-9]*) echo "database query did not return an unsigned integer: $value" >&2; return 1 ;;
esac
printf '%s' "$value"
}
save_logs() {
compose logs --no-color > "$REPORT_DIR/compose.log" 2>&1 || true
}
copy_reports() {
if ! reports_container=$(compose ps -aq reports 2>/dev/null); then
echo "could not locate the stress reports container; preserving Docker volumes for recovery" >&2
return 1
fi
if [ -z "$reports_container" ]; then
echo "stress reports container does not exist; preserving Docker volumes for recovery" >&2
return 1
fi
if ! MSYS_NO_PATHCONV=1 docker cp "$reports_container:/reports/." "$REPORT_DIR_DOCKER"; then
echo "could not copy stress reports; preserving Docker volumes for recovery" >&2
return 1
fi
}
cleanup() {
status=$1
trap - 0 2 15
reports_copied=true
if ! copy_reports; then
reports_copied=false
status=1
fi
save_logs
if [ "$reports_copied" = true ]; then
compose down --volumes --remove-orphans >/dev/null 2>&1 || true
else
compose down --remove-orphans >/dev/null 2>&1 || true
fi
exit "$status"
}
trap 'cleanup $?' 0
trap 'cleanup 130' 2
trap 'cleanup 143' 15
wait_for_streams() {
condition=$1
minimum=$2
timeout=$3
label=$4
watch_pid=${5:-}
deadline=$(( $(date +%s) + timeout ))
while :; do
if [ -n "$watch_pid" ] && ! kill -0 "$watch_pid" 2>/dev/null; then
echo "$label stopped because its health probe exited early" >&2
return 1
fi
current=$(db_integer "SELECT count(*) FROM stream WHERE $condition")
if [ "$current" -ge "$minimum" ]; then
printf '%s reached %s streams (required %s)\n' "$label" "$current" "$minimum"
return 0
fi
if [ "$(date +%s)" -ge "$deadline" ]; then
echo "$label reached only $current of required $minimum streams" >&2
return 1
fi
sleep 2
done
}
wait_for_zero() {
condition=$1
timeout=$2
label=$3
watch_pid=${4:-}
deadline=$(( $(date +%s) + timeout ))
while :; do
if [ -n "$watch_pid" ] && ! kill -0 "$watch_pid" 2>/dev/null; then
echo "$label stopped because its health probe exited early" >&2
return 1
fi
current=$(db_integer "SELECT count(*) FROM stream WHERE $condition")
if [ "$current" -eq 0 ]; then
printf '%s drained\n' "$label"
return 0
fi
if [ "$(date +%s)" -ge "$deadline" ]; then
echo "$label still has $current streams" >&2
return 1
fi
sleep 2
done
}
check_rate() {
phase=$1
streams=$2
elapsed=$3
if [ "$elapsed" -lt 1 ]; then elapsed=1; fi
if [ "$streams" -lt $((MIN_STREAMS_PER_SECOND * elapsed)) ]; then
echo "$phase achieved less than $MIN_STREAMS_PER_SECOND completed TCP streams/s ($streams in ${elapsed}s)" >&2
return 1
fi
printf '%s completed %s streams in %ss (at least %s streams/s)\n' \
"$phase" "$streams" "$elapsed" "$MIN_STREAMS_PER_SECOND"
}
load_phase_elapsed() {
phase=$1
compose exec -T reports python -c '
import glob, json, math, sys
paths = glob.glob(f"/reports/{sys.argv[1]}-*.json")
if not paths:
raise SystemExit(f"no generator reports found for phase {sys.argv[1]}")
reports = [json.load(open(path, encoding="utf-8")) for path in paths]
reports = [report for report in reports if "startedAtUnixSeconds" in report]
if not reports:
raise SystemExit(f"no timed generator reports found for phase {sys.argv[1]}")
elapsed = max(report["finishedAtUnixSeconds"] for report in reports) \
- min(report["startedAtUnixSeconds"] for report in reports)
print(max(1, math.ceil(elapsed)))
' "$phase" | tr -d '\r'
}
emit_ledger() {
stream_condition=$1
compose exec -T db psql -v ON_ERROR_STOP=1 -U packmate -d packmate -At \
-F "$(printf '\t')" -c "
WITH payloads AS (
SELECT s.id AS stream_id,
string_agg(p.content, ''::bytea ORDER BY p.id) FILTER (WHERE p.incoming) AS incoming_content,
string_agg(p.content, ''::bytea ORDER BY p.id) FILTER (WHERE NOT p.incoming) AS outgoing_content,
bool_and(COALESCE(
CASE WHEN p.incoming
THEN p.destination_ip = '$TARGET_IP' AND p.source_ip <> '$TARGET_IP'
ELSE p.source_ip = '$TARGET_IP' AND p.destination_ip <> '$TARGET_IP'
END,
false
)) AS ips_valid,
count(p.id) = s.packets_count
AND sum(octet_length(p.content)) = s.size_bytes
AND s.service_id = 18080
AND s.protocol = 0
AND (s.end_timestamp = 0 OR s.start_timestamp <= s.end_timestamp)
AS metadata_valid
FROM stream s
JOIN packet p ON p.stream_id = s.id
WHERE $stream_condition
GROUP BY s.id, s.packets_count, s.size_bytes, s.service_id, s.protocol,
s.start_timestamp, s.end_timestamp
)
SELECT stream_id,
encode(substring(incoming_content FROM 5 FOR 24), 'hex'),
octet_length(incoming_content),
md5(incoming_content),
octet_length(outgoing_content),
md5(outgoing_content),
ips_valid,
metadata_valid
FROM payloads
ORDER BY stream_id"
}
count_invalid_matches() {
stream_condition=$1
db_integer "
SELECT count(*)
FROM (
SELECT s.id
FROM stream s
JOIN packet p ON p.stream_id = s.id
LEFT JOIN found_pattern fp ON fp.packet_id = p.id
WHERE $stream_condition
GROUP BY s.id
HAVING count(fp.id) <> 2
OR count(fp.id) FILTER (WHERE p.incoming) <> 1
OR count(fp.id) FILTER (WHERE NOT p.incoming) <> 1
OR bool_and(
substring(
p.content
FROM fp.start_position + 1
FOR fp.end_position - fp.start_position + 1
) = convert_to('PMSTRESS', 'UTF8')
AND COALESCE(fp.search_target, 'CONTENT') = 'CONTENT'
AND fp.pattern_id = (SELECT id FROM pattern WHERE name = 'network-stress-marker')
) FILTER (WHERE fp.id IS NOT NULL) IS NOT TRUE
) invalid"
}
verify_complete_ledger() {
condition=$1
run_id=$2
total=$3
expected=$4
report=$5
subset=${6:-false}
subset_arg=""
ledger_file="$REPORT_DIR/$report.tsv"
if [ "$subset" = true ]; then subset_arg="--allow-subset"; fi
emit_ledger "$condition" > "$ledger_file"
# shellcheck disable=SC2086
compose run --rm -T --no-deps generator \
python /stress/verify_ledger.py \
--run-id "$run_id" \
--total-streams "$total" \
--expected-streams "$expected" \
--shards "$SHARDS" \
--frames "$FRAMES" \
--payload-bytes "$PAYLOAD_BYTES" \
--report "/reports/$report" \
$subset_arg < "$ledger_file"
rm -f "$ledger_file"
}
run_load_phase() {
phase=$1
total=$2
run_id=$3
abort_file=$4
minimum_duration=$5
base=$((total / SHARDS))
remainder=$((total % SHARDS))
pids=""
shard=0
offset=0
while [ "$shard" -lt "$SHARDS" ]; do
count=$base
if [ "$shard" -lt "$remainder" ]; then count=$((count + 1)); fi
if [ "$count" -gt 0 ]; then
compose run --rm --no-deps generator python /stress/network_load.py \
--streams "$count" \
--frames "$FRAMES" \
--payload-bytes "$PAYLOAD_BYTES" \
--concurrency "$CONCURRENCY" \
--run-id "$run_id" \
--shard "$shard" \
--stream-offset "$offset" \
--abort-file "$abort_file" \
--minimum-duration-seconds "$minimum_duration" \
--report "/reports/$phase-$shard.json" &
pids="$pids $!"
fi
offset=$((offset + count))
shard=$((shard + 1))
done
failed=0
for pid in $pids; do
if ! wait "$pid"; then failed=1; fi
done
return "$failed"
}
echo "Starting isolated network stress project $PROJECT; reports: $REPORT_DIR"
compose up -d --build db target packmate reports
compose build generator
compose exec -T reports sh -c 'rm -f /reports/*'
compose run --rm --no-deps generator python /stress/api_control.py prepare --timeout 300
db_df=$(compose exec -T db df -PkT /var/lib/postgresql/data) || exit $?
db_available_kb=$(printf '%s\n' "$db_df" | awk 'NR == 2 {print $5}' | tr -d '\r')
db_storage_type=$(printf '%s\n' "$db_df" | awk 'NR == 2 {print $2}' | tr -d '\r')
require_positive "$db_available_kb" "database available storage"
estimated_raw_bytes=$(( (BASELINE_STREAMS + LIVE_STREAMS) * FRAMES * (PAYLOAD_BYTES + 4) * 2 ))
required_storage_bytes=$((estimated_raw_bytes * STORAGE_MULTIPLIER))
available_storage_bytes=$((db_available_kb * 1024))
if [ "$available_storage_bytes" -lt "$required_storage_bytes" ]; then
echo "database volume has $available_storage_bytes bytes free; stress preflight requires $required_storage_bytes" >&2
exit 1
fi
cat > "$REPORT_DIR/effective-config.json" <<EOF
{
"project": "$PROJECT",
"runId": "$RUN_LABEL",
"baselineStreams": $BASELINE_STREAMS,
"concurrentStreams": $LIVE_STREAMS,
"shards": $SHARDS,
"concurrencyPerShard": $CONCURRENCY,
"framesPerStream": $FRAMES,
"payloadBytes": $PAYLOAD_BYTES,
"minimumLiveDurationSeconds": $MIN_LIVE_DURATION_SECONDS,
"maximumCaptureQueue": $MAX_CAPTURE_QUEUE,
"captureWorkers": $CAPTURE_WORKERS,
"captureQueueCapacity": $CAPTURE_QUEUE_CAPACITY,
"captureBufferSizeBytes": $CAPTURE_BUFFER_SIZE_BYTES,
"tcpTimeoutSeconds": $TCP_TIMEOUT_SECONDS,
"udpTimeoutSeconds": $UDP_TIMEOUT_SECONDS,
"minimumStreamsPerSecond": $MIN_STREAMS_PER_SECOND,
"minimumCleanupStreamsPerSecond": $MIN_CLEANUP_STREAMS_PER_SECOND,
"cleanupIntervalSeconds": $PACKMATE_NETWORK_STRESS_CLEANUP_INTERVAL_SECONDS,
"cleanupRetentionMinutes": $PACKMATE_NETWORK_STRESS_CLEANUP_THRESHOLD_MINUTES,
"targetIp": "$TARGET_IP",
"databaseVolumeFreeBytesAtStart": $available_storage_bytes,
"databaseStorageType": "$db_storage_type"
}
EOF
baseline_started=$(date +%s)
compose run --rm --no-deps generator python /stress/api_control.py monitor \
--stop-file /reports/baseline-load.done \
--abort-file /reports/baseline-load.abort \
--max-queue "$MAX_CAPTURE_QUEUE" \
--report /reports/baseline-capture-monitor.json &
baseline_monitor_pid=$!
baseline_load_status=0
if ! run_load_phase baseline "$BASELINE_STREAMS" 1 /reports/baseline-load.abort 0; then
baseline_load_status=1
fi
compose exec -T reports touch /reports/baseline-load.done
baseline_monitor_status=0
if ! wait "$baseline_monitor_pid"; then baseline_monitor_status=1; fi
if [ "$baseline_load_status" -ne 0 ] || [ "$baseline_monitor_status" -ne 0 ]; then
echo "baseline traffic or continuous capture monitor failed" >&2
exit 1
fi
baseline_load_elapsed=$(load_phase_elapsed baseline)
require_positive "$baseline_load_elapsed" "baseline generator elapsed time"
check_rate "baseline generators" "$BASELINE_STREAMS" "$baseline_load_elapsed"
compose run --rm --no-deps generator python /stress/api_control.py metrics \
--max-queue "$MAX_CAPTURE_QUEUE" \
--report /reports/pcap-metrics-after-baseline-load.json
baseline_drain_started=$(date +%s)
wait_for_streams "end_timestamp > 0 AND favorite = false" "$BASELINE_STREAMS" "$DRAIN_TIMEOUT" "baseline network ingest"
baseline_drain_seconds=$(( $(date +%s) - baseline_drain_started ))
if [ "$baseline_drain_seconds" -gt "$MAX_INGEST_DRAIN_SECONDS" ]; then
echo "baseline ingest queue needed ${baseline_drain_seconds}s to drain" >&2
exit 1
fi
compose run --rm --no-deps generator python /stress/api_control.py metrics \
--require-empty-queue \
--wait-seconds "$DRAIN_TIMEOUT" \
--report /reports/pcap-metrics-before-cleanup.json
compose run --rm --no-deps generator python /stress/api_control.py verify \
--run-id 1 \
--sample-size 200 \
--frames "$FRAMES" \
--payload-bytes "$PAYLOAD_BYTES" \
--report /reports/baseline-payload-verification.json
baseline_captured=$(db_integer "SELECT count(*) FROM stream")
baseline_packets=$(db_integer "SELECT count(*) FROM packet")
baseline_matches=$(db_integer "SELECT count(*) FROM found_pattern")
baseline_pattern_links=$(db_integer "SELECT count(*) FROM stream_found_patterns")
invalid_baseline_matches=$(count_invalid_matches "true")
if [ "$baseline_captured" -ne "$BASELINE_STREAMS" ] \
|| [ "$baseline_matches" -ne $((BASELINE_STREAMS * 2)) ] \
|| [ "$baseline_pattern_links" -ne "$BASELINE_STREAMS" ] \
|| [ "$invalid_baseline_matches" -ne 0 ]; then
echo "invalid baseline counts: streams=$baseline_captured matches=$baseline_matches links=$baseline_pattern_links invalidMatches=$invalid_baseline_matches" >&2
exit 1
fi
verify_complete_ledger "true" 1 "$BASELINE_STREAMS" "$BASELINE_STREAMS" baseline-full-ledger.json
db_scalar "UPDATE stream SET favorite = true WHERE id IN (SELECT id FROM stream ORDER BY id LIMIT $FAVORITE_STREAMS)" >/dev/null
favorite_count=$(db_integer "SELECT count(*) FROM stream WHERE favorite")
if [ "$favorite_count" -ne "$FAVORITE_STREAMS" ]; then
echo "could not mark $FAVORITE_STREAMS baseline streams as favorite" >&2
exit 1
fi
db_bytes_before=$(db_integer "SELECT pg_database_size(current_database())")
wal_start=$(db_scalar "SELECT pg_current_wal_lsn()")
db_scalar "UPDATE stream SET end_timestamp = 0, persisted_at = 0" >/dev/null
eligible_baseline=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp = 0 AND favorite = false")
expected_eligible=$((BASELINE_STREAMS - FAVORITE_STREAMS))
if [ "$eligible_baseline" -ne "$expected_eligible" ]; then
echo "could not create the complete eligible baseline: eligible=$eligible_baseline expected=$expected_eligible" >&2
exit 1
fi
cleanup_measurement_started_at=$(date +%s)
printf 'unix_seconds,eligible_baseline_streams,committed_live_streams\n' > "$REPORT_DIR/cleanup-overlap.csv"
printf '%s,%s,0\n' "$cleanup_measurement_started_at" "$eligible_baseline" >> "$REPORT_DIR/cleanup-overlap.csv"
PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED=true
export PACKMATE_NETWORK_STRESS_CLEANUP_ENABLED
compose up -d --no-deps --force-recreate packmate
compose run --rm --no-deps generator python /stress/api_control.py prepare --timeout 300
compose run --rm --no-deps generator python /stress/api_control.py probe \
--requests 20000 \
--pause 0.1 \
--min-requests 100 \
--stop-file /reports/workload.done \
--abort-file /reports/live-load.abort \
--max-queue "$MAX_CAPTURE_QUEUE" \
--max-p99-ms "$MAX_API_P99_MS" \
--report /reports/api-probe.json &
probe_pid=$!
run_load_phase live "$LIVE_STREAMS" 2 /reports/live-load.abort "$MIN_LIVE_DURATION_SECONDS" &
live_pid=$!
overlap_observed=false
cleanup_completed_at=0
previous_old=$eligible_baseline
previous_live=0
while kill -0 "$live_pid" 2>/dev/null; do
sample_time=$(date +%s)
remaining_old=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp = 0 AND favorite = false")
committed_live=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp > 0 AND favorite = false")
printf '%s,%s,%s\n' "$sample_time" "$remaining_old" "$committed_live" >> "$REPORT_DIR/cleanup-overlap.csv"
if [ "$remaining_old" -lt "$previous_old" ] \
&& [ "$committed_live" -gt "$previous_live" ] \
&& [ "$previous_live" -gt 0 ]; then
overlap_observed=true
fi
if [ "$remaining_old" -eq 0 ] && [ "$cleanup_completed_at" -eq 0 ]; then
cleanup_completed_at=$sample_time
fi
if ! kill -0 "$probe_pid" 2>/dev/null; then
echo "API/capture probe exited while concurrent traffic was active" >&2
compose exec -T reports touch /reports/live-load.abort
fi
previous_old=$remaining_old
previous_live=$committed_live
sleep 1
done
live_load_status=0
if ! wait "$live_pid"; then live_load_status=1; fi
if [ "$live_load_status" -ne 0 ]; then
compose exec -T reports touch /reports/workload.done
wait "$probe_pid" || true
echo "concurrent traffic was aborted or a generator failed" >&2
exit 1
fi
if [ "$overlap_observed" != true ]; then
compose exec -T reports touch /reports/workload.done
wait "$probe_pid" || true
echo "cleanup did not overlap committed live network traffic" >&2
exit 1
fi
live_load_elapsed=$(load_phase_elapsed live)
require_positive "$live_load_elapsed" "live generator elapsed time"
check_rate "live generators" "$LIVE_STREAMS" "$live_load_elapsed"
live_drain_started=$(date +%s)
wait_for_streams "end_timestamp > 0 AND favorite = false" "$LIVE_STREAMS" "$DRAIN_TIMEOUT" "concurrent network ingest" "$probe_pid"
live_drain_seconds=$(( $(date +%s) - live_drain_started ))
if [ "$live_drain_seconds" -gt "$MAX_INGEST_DRAIN_SECONDS" ]; then
echo "live ingest queue needed ${live_drain_seconds}s to drain" >&2
exit 1
fi
wait_for_zero "end_timestamp = 0 AND favorite = false" "$CLEANUP_TIMEOUT" "eligible cleanup backlog" "$probe_pid"
if [ "$cleanup_completed_at" -eq 0 ]; then cleanup_completed_at=$(date +%s); fi
cleanup_elapsed_seconds=$((cleanup_completed_at - cleanup_measurement_started_at))
if [ "$cleanup_elapsed_seconds" -lt 1 ]; then cleanup_elapsed_seconds=1; fi
cleanup_streams_per_second=$((eligible_baseline / cleanup_elapsed_seconds))
if [ "$cleanup_streams_per_second" -lt "$MIN_CLEANUP_STREAMS_PER_SECOND" ]; then
echo "cleanup sustained only $cleanup_streams_per_second streams/s; required $MIN_CLEANUP_STREAMS_PER_SECOND" >&2
exit 1
fi
compose run --rm -T --no-deps generator sh -c 'touch /reports/workload.done'
wait "$probe_pid"
compose run --rm --no-deps generator python /stress/api_control.py verify \
--run-id 2 \
--sample-size 200 \
--frames "$FRAMES" \
--payload-bytes "$PAYLOAD_BYTES" \
--report /reports/concurrent-payload-verification.json
verify_complete_ledger "s.favorite = false" 2 "$LIVE_STREAMS" "$LIVE_STREAMS" concurrent-full-ledger.json
verify_complete_ledger "s.favorite = true" 1 "$BASELINE_STREAMS" "$FAVORITE_STREAMS" favorite-full-ledger.json true
compose run --rm --no-deps generator python /stress/api_control.py metrics \
--require-empty-queue \
--wait-seconds "$DRAIN_TIMEOUT" \
--report /reports/pcap-metrics.json
orphan_packets=$(db_integer "SELECT count(*) FROM packet p LEFT JOIN stream s ON s.id=p.stream_id WHERE s.id IS NULL")
orphan_matches=$(db_integer "SELECT count(*) FROM found_pattern fp LEFT JOIN packet p ON p.id=fp.packet_id WHERE p.id IS NULL")
orphan_pattern_links=$(db_integer "SELECT count(*) FROM stream_found_patterns sfp LEFT JOIN stream s ON s.id=sfp.stream_id LEFT JOIN pattern pt ON pt.id=sfp.pattern_id WHERE s.id IS NULL OR pt.id IS NULL")
cleanup_pending=$(db_integer "SELECT count(*) FROM stream WHERE cleanup_pending")
final_matches=$(db_integer "SELECT count(*) FROM found_pattern")
final_pattern_links=$(db_integer "SELECT count(*) FROM stream_found_patterns")
invalid_final_matches=$(count_invalid_matches "true")
if [ "$orphan_packets" -ne 0 ] \
|| [ "$orphan_matches" -ne 0 ] \
|| [ "$orphan_pattern_links" -ne 0 ] \
|| [ "$cleanup_pending" -ne 0 ] \
|| [ "$final_matches" -ne $(((LIVE_STREAMS + FAVORITE_STREAMS) * 2)) ] \
|| [ "$final_pattern_links" -ne $((LIVE_STREAMS + FAVORITE_STREAMS)) ] \
|| [ "$invalid_final_matches" -ne 0 ]; then
echo "invalid final state: orphan_packets=$orphan_packets orphan_matches=$orphan_matches orphan_links=$orphan_pattern_links cleanup_pending=$cleanup_pending matches=$final_matches links=$final_pattern_links invalidMatches=$invalid_final_matches" >&2
exit 1
fi
live_captured=$(db_integer "SELECT count(*) FROM stream WHERE end_timestamp > 0 AND favorite = false")
favorite_captured=$(db_integer "SELECT count(*) FROM stream WHERE favorite")
final_packets=$(db_integer "SELECT count(*) FROM packet")
db_bytes_after=$(db_integer "SELECT pg_database_size(current_database())")
wal_bytes=$(db_integer "SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '$wal_start')::bigint")
elapsed=$(( $(date +%s) - baseline_started ))
cat > "$REPORT_DIR/summary.json" <<EOF
{
"baselineRequestedStreams": $BASELINE_STREAMS,
"baselineCapturedStreams": $baseline_captured,
"baselineCapturedPackets": $baseline_packets,
"baselineMatches": $baseline_matches,
"baselinePatternLinks": $baseline_pattern_links,
"concurrentRequestedStreams": $LIVE_STREAMS,
"concurrentCapturedStreams": $live_captured,
"favoriteStreamsPreserved": $favorite_captured,
"finalPackets": $final_packets,
"finalMatches": $final_matches,
"finalPatternLinks": $final_pattern_links,
"shards": $SHARDS,
"framesPerStream": $FRAMES,
"payloadBytes": $PAYLOAD_BYTES,
"captureWorkers": $CAPTURE_WORKERS,
"captureQueueCapacity": $CAPTURE_QUEUE_CAPACITY,
"captureBufferSizeBytes": $CAPTURE_BUFFER_SIZE_BYTES,
"tcpTimeoutSeconds": $TCP_TIMEOUT_SECONDS,
"elapsedSeconds": $elapsed,
"baselineLoadSeconds": $baseline_load_elapsed,
"baselineIngestDrainSeconds": $baseline_drain_seconds,
"liveLoadSeconds": $live_load_elapsed,
"liveIngestDrainSeconds": $live_drain_seconds,
"minimumStreamsPerSecond": $MIN_STREAMS_PER_SECOND,
"cleanupOverlapObserved": $overlap_observed,
"cleanupElapsedSeconds": $cleanup_elapsed_seconds,
"cleanupStreamsPerSecond": $cleanup_streams_per_second,
"minimumCleanupStreamsPerSecond": $MIN_CLEANUP_STREAMS_PER_SECOND,
"walBytesDuringCleanup": $wal_bytes,
"databaseBytesBeforeCleanup": $db_bytes_before,
"databaseBytesAfterCleanup": $db_bytes_after,
"databaseVolumeFreeBytesAtStart": $available_storage_bytes,
"databaseStorageType": "$db_storage_type",
"estimatedRawTrafficBytes": $estimated_raw_bytes,
"requiredStoragePreflightBytes": $required_storage_bytes,
"orphanPackets": $orphan_packets,
"orphanMatches": $orphan_matches,
"orphanPatternLinks": $orphan_pattern_links,
"cleanupPending": $cleanup_pending
}
EOF
cat "$REPORT_DIR/summary.json"
echo "Network capture and cleanup stress PASSED"
Reference in New Issue View Git Blame Copy Permalink