#!/usr/bin/env bash # Live M5 smoke test. Run from repo root: # bash scripts/m5_smoke.sh # # Walks through the 6 scenarios in M5_VERIFICATION.md: # # Step 2 — happy path: 1 alert via WS → 2 deliveries (Alice fcm + telegram) # Step 3 — 5 alerts, mixed severity + 30% dedupe → 10 deliveries # Step 4 — bad API key (correct format, wrong secret) → 0 deliveries, # ws_messages_total{result="bad_signature"} ticks # Step 5 — per-IP concurrency cap (35 conns, cap=32) → 32 ok + 3 rejected, # connection_rejected_total{transport="ws"} = 3 # Step 6 — live tail: subscribe then send → 1 frame arrives, # tail_subscribers = 1 while connected # Step 7 — live tail company filter: subscribe to globex-002 only → # acme alerts filtered out, globex alert passes # # The script assumes the loadgen-ws binary is built at /tmp/loadgen-ws # (run `cd loadgen && go build -o /tmp/loadgen-ws ./cmd/ws`). The # per-IP cap test binary and tail driver are auto-built on first run; # they live in /tmp and are reused on subsequent runs. # # Exit code is the number of failed checks. set -e cd "$(dirname "$0")/.." PG="docker exec -i broad-announce-postgres-1 psql -U ba -d ba -A -t" INGESTD_METRICS=http://localhost:8800/metrics SRC_ACME=acme-001:prom-prod:s3cret-acme SRC_GLOBEX=globex-002:grafana:s3cret-globex WS_INGEST=ws://localhost:8800/v1/ingest/ws WS_TAIL=ws://localhost:8800/v1/tail/ws TAIL_TOKEN=tail-dev-token-please-change-in-prod fails=0 pass() { echo " ✅ $*"; } fail() { echo " ❌ $*"; fails=$((fails+1)); } reset_state() { $PG -c "UPDATE individuals SET telegram_chat_id = NULL, telegram_user_id = NULL, telegram_invite_code = 'acme-bob-002' WHERE id = 'ind-acme-002';" >/dev/null $PG -c "UPDATE individuals SET telegram_chat_id = NULL, telegram_user_id = NULL WHERE id = 'ind-acme-003';" >/dev/null $PG -c "UPDATE subscriptions SET min_severity = 'critical' WHERE individual_id = 'ind-acme-002' AND source_id = 'prom-prod';" >/dev/null $PG -c "TRUNCATE deliveries;" >/dev/null curl -sS -X POST http://localhost:8830/admin/reset >/dev/null docker compose restart telegramd >/dev/null for i in 1 2 3 4 5 6 7 8 9 10; do if curl -sS http://localhost:8822/health 2>/dev/null | grep -q '"status":"ok"'; then sleep 1 break fi sleep 1 done } ws_counter() { # $1 = result label curl -sS "$INGESTD_METRICS" 2>/dev/null | \ grep -E "^ba_ingestd_ws_messages_total\{result=\"$1\"" | \ awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1 } ws_conn() { # $1 = state label curl -sS "$INGESTD_METRICS" 2>/dev/null | \ grep -E "^ba_ingestd_ws_connections_total\{state=\"$1\"" | \ awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1 } ws_rej() { curl -sS "$INGESTD_METRICS" 2>/dev/null | \ grep -E '^ba_ingestd_connection_rejected_total\{[^}]*transport="ws"[^}]*\}' | \ awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1 } tail_gauge() { curl -sS "$INGESTD_METRICS" 2>/dev/null | \ grep -E '^ba_ingestd_tail_subscribers\{' | \ awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1 } # ── Setup: build loadgen-ws + per-IP-cap test binary + tail driver ── mkdir -p /tmp/m5_smoke if [[ ! -x /tmp/loadgen-ws ]]; then echo "▸ Building /tmp/loadgen-ws" (cd loadgen && CGO_ENABLED=0 go build -o /tmp/loadgen-ws ./cmd/ws) fi if [[ ! -x /tmp/m5-tail-test ]]; then echo "▸ Building /tmp/m5-tail-test" CGO_ENABLED=0 go build -o /tmp/m5-tail-test ./loadgen/cmd/m5drivers/tail fi if [[ ! -x /tmp/m5-perip-test ]]; then echo "▸ Building /tmp/m5-perip-test" cat > /tmp/m5_smoke/perip_test.go <<'GO' package main import ( "flag" "fmt" "sync" "sync/atomic" "git3.techno-world.net/lrosales/broad-announce/internal/wsclient" ) func main() { wsURL := flag.String("ws", "ws://localhost:8800/v1/ingest/ws", "ws endpoint") apiKey := flag.String("key", "acme-001:prom-prod:s3cret-acme", "api key") cap := flag.Int("cap", 35, "number of concurrent WS conns to attempt (cap=32 should reject 3)") flag.Parse() var ok, rejected atomic.Int64 var wg sync.WaitGroup for i := 0; i < *cap; i++ { wg.Add(1) go func() { defer wg.Done() c, err := wsclient.Connect(wsclient.Config{URL: *wsURL, APIKey: *apiKey, DialTimeout: 5e9}) if err != nil { rejected.Add(1) return } ok.Add(1) defer c.Close() }() } wg.Wait() fmt.Printf("ok=%d rejected=%d (cap=32 → ok<=32, rejected>=3 if cap=35)\n", ok.Load(), rejected.Load()) } GO mkdir -p ./scripts/m5_smoke_tmp cp /tmp/m5_smoke/perip_test.go ./scripts/m5_smoke_tmp/main.go CGO_ENABLED=0 go build -o /tmp/m5-perip-test ./scripts/m5_smoke_tmp rm -rf ./scripts/m5_smoke_tmp fi # ───────────────────────────────────────────────────────────────── echo "── M5 smoke — WebSocket ingest + live tail + per-IP cap ──" echo "" reset_state # ── Step 2: happy path ──────────────────────────────────────── echo "── Step 2: 1 alert via WS → 2 deliveries (Alice fcm + telegram) ──" acme_recv_before=$(ws_counter received) acme_acc_before=$(ws_counter accepted) deliveries_before=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n') acme_recv_before=${acme_recv_before:-0} acme_acc_before=${acme_acc_before:-0} deliveries_before=${deliveries_before:-0} /tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 1 --rate 1 2>&1 | tail -3 sleep 2 acme_recv_after=$(ws_counter received) acme_acc_after=$(ws_counter accepted) deliveries_after=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n') acme_recv_after=${acme_recv_after:-0} acme_acc_after=${acme_acc_after:-0} deliveries_after=${deliveries_after:-0} recv_delta=$((acme_recv_after - acme_recv_before)) acc_delta=$((acme_acc_after - acme_acc_before)) del_delta=$((deliveries_after - deliveries_before)) if [[ $recv_delta -ge 1 ]]; then pass "ws_messages_total{result=\"received\"} +$recv_delta" else fail "ws_messages_total{result=\"received\"} delta was $recv_delta (expected ≥1)" fi if [[ $acc_delta -ge 1 ]]; then pass "ws_messages_total{result=\"accepted\"} +$acc_delta" else fail "ws_messages_total{result=\"accepted\"} delta was $acc_delta (expected ≥1)" fi if [[ $del_delta -ge 2 ]]; then pass "deliveries +$del_delta (Alice fcm + telegram)" else fail "deliveries delta was $del_delta (expected ≥2)" fi # ── Step 3: 5 alerts, mixed severity + 30% dedupe ─────────── echo "" echo "── Step 3: 5 alerts, mixed severity + 30% dedupe → 10 deliveries ──" reset_state acme_recv_before=$(ws_counter received) acme_acc_before=$(ws_counter accepted) deliveries_before=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n') # Send 5 alerts at a steady rate. Default profile is # normal which uses 30% dedupe (i.e. ~2 of 5 alerts share # a dedupe_key), so we expect 3 new accepts and 2 dupes. /tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 5 --rate 5 2>&1 | tail -3 sleep 3 acme_recv_after=$(ws_counter received) acme_acc_after=$(ws_counter accepted) deliveries_after=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n') recv_delta=$((acme_recv_after - acme_recv_before)) acc_delta=$((acme_acc_after - acme_acc_before)) del_delta=$((deliveries_after - deliveries_before)) if [[ $recv_delta -eq 5 ]]; then pass "ws_messages_total{result=\"received\"} +$recv_delta" else fail "ws_messages_total{result=\"received\"} delta was $recv_delta (expected 5)" fi if [[ $acc_delta -ge 3 ]]; then pass "ws_messages_total{result=\"accepted\"} +$acc_delta (≥3 of 5 are new; rest deduped)" else fail "ws_messages_total{result=\"accepted\"} delta was $acc_delta (expected ≥3)" fi if [[ $del_delta -ge 8 ]]; then pass "deliveries +$del_delta (≥1 per non-deduped alert × 2 channels)" else fail "deliveries delta was $del_delta (expected ≥8)" fi # ── Step 4: bad API key ───────────────────────────────────── echo "" echo "── Step 4: bad API key (wrong secret) → 0 deliveries, bad_signature counter ticks ──" reset_state bad_before=$(ws_counter bad_signature) deliveries_before=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n') # Use the right company:source (so auth passes) but wrong HMAC. cat > /tmp/m5_smoke/badkey.go <<'GO' package main import ( "crypto/hmac" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "time" "github.com/gorilla/websocket" ) func main() { conn, _, err := websocket.DefaultDialer.Dial("ws://localhost:8800/v1/ingest/ws", nil) if err != nil { fmt.Println("dial err:", err); return } defer conn.Close() type Auth struct { APIKey string `json:"api_key"` } if err := conn.WriteJSON(Auth{APIKey: "acme-001:prom-prod:s3cret-acme"}); err != nil { fmt.Println("auth err:", err); return } conn.SetReadDeadline(time.Now().Add(2*time.Second)) _, authReply, err := conn.ReadMessage() if err != nil { fmt.Println("auth reply err:", err); return } fmt.Println("auth reply:", string(authReply)) body := map[string]any{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"smoke-badkey","labels":map[string]string{"instance":"smoke-1"}} bb, _ := json.Marshal(body) // Sign with a wrong key — should fail HMAC verify. mac := hmac.New(sha256.New, []byte("WRONG-SECRET")) mac.Write(bb) hexMac := hex.EncodeToString(mac.Sum(nil)) sig := "t=1700000000,v1=" + hexMac env := map[string]any{ "alert": json.RawMessage(bb), "auth": sig, } if err := conn.WriteJSON(env); err != nil { fmt.Println("write err:", err); return } conn.SetReadDeadline(time.Now().Add(2*time.Second)) _, ack2, err := conn.ReadMessage() if err != nil { fmt.Println("ack2 err:", err); return } fmt.Println("ack2:", string(ack2)) } GO mkdir -p ./scripts/m5_smoke_tmp cp /tmp/m5_smoke/badkey.go ./scripts/m5_smoke_tmp/main.go CGO_ENABLED=0 go build -o /tmp/m5-badkey ./scripts/m5_smoke_tmp rm -rf ./scripts/m5_smoke_tmp /tmp/m5-badkey 2>&1 | tail -5 sleep 2 bad_after=$(ws_counter bad_signature) deliveries_after=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n') bad_delta=$((bad_after - bad_before)) del_delta=$((deliveries_after - deliveries_before)) if [[ $bad_delta -ge 1 ]]; then pass "ws_messages_total{result=\"bad_signature\"} +$bad_delta" else fail "ws_messages_total{result=\"bad_signature\"} delta was $bad_delta (expected ≥1)" fi if [[ $del_delta -eq 0 ]]; then pass "no deliveries created" else fail "deliveries delta was $del_delta (expected 0)" fi # ── Step 5: per-IP concurrency cap ───────────────────────── echo "" echo "── Step 5: per-IP cap (35 conns, cap=32) → 32 ok + 3 rejected ──" # Drain any residual per-IP count from prior tests. The per-IP # counter decrements on TCP close; some kernels buffer the # FIN/ACK and we can race the next test. A 5-second wait is # enough for the janitor to forget any stragglers (idle > 1s # in the test) and for the kernel to reap closed sockets. sleep 5 rej_before=$(ws_rej) perip_out=$(/tmp/m5-perip-test --cap 35 2>&1 | tail -1) echo " perip_test: $perip_out" sleep 1 rej_after=$(ws_rej) rej_delta=$((rej_after - rej_before)) if echo "$perip_out" | grep -qE "ok=3[0-2] rejected=[2-3]"; then pass "perip_test reported 30-32 ok, 2-3 rejected (cap=32 held)" else fail "perip_test output unexpected: $perip_out" fi if [[ $rej_delta -ge 2 && $rej_delta -le 3 ]]; then pass "connection_rejected_total{transport=\"ws\"} +$rej_delta" else fail "connection_rejected_total{transport=\"ws\"} delta was $rej_delta (expected 2-3)" fi sleep 1 # ── Step 6: live tail, no filter ─────────────────────────── echo "" echo "── Step 6: live tail (no filter) — subscribe then send 1 alert → 1 frame arrives ──" tail_before=$(tail_gauge) /tmp/m5-tail-test --target "$WS_TAIL" --token "$TAIL_TOKEN" --duration 6s 2>&1 > /tmp/m5_tail1.log & TAIL_PID=$! sleep 2 # let the tail subscribe tail_mid=$(tail_gauge) if [[ $tail_mid -ge 1 ]]; then pass "tail_subscribers = $tail_mid while connected" else fail "tail_subscribers was $tail_mid (expected ≥1)" fi /tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 1 --rate 1 2>&1 | tail -1 wait $TAIL_PID 2>/dev/null if grep -q "^FRAME:" /tmp/m5_tail1.log; then pass "tail received at least 1 frame" else fail "tail received 0 frames (log: $(cat /tmp/m5_tail1.log))" fi tail_after=$(tail_gauge) if [[ $tail_after -eq 0 ]]; then pass "tail_subscribers back to 0 after disconnect" else fail "tail_subscribers was $tail_after after disconnect (expected 0)" fi # ── Step 7: live tail, company filter ───────────────────── echo "" echo "── Step 7: live tail (company=globex-002) — acme filtered, globex passes ──" /tmp/m5-tail-test --target "$WS_TAIL" --token "$TAIL_TOKEN" --company globex-002 --duration 6s 2>&1 > /tmp/m5_tail2.log & TAIL_PID=$! sleep 2 # Send acme — should NOT reach the tail /tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 1 --rate 1 2>&1 | tail -1 sleep 1 # Send globex — should reach the tail /tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_GLOBEX" --count 1 --rate 1 2>&1 | tail -1 wait $TAIL_PID 2>/dev/null # Count frames per company acme_frames=$(grep -c "company_id\":\"acme-001" /tmp/m5_tail2.log || true) globex_frames=$(grep -c "company_id\":\"globex-002" /tmp/m5_tail2.log || true) if [[ $acme_frames -eq 0 ]]; then pass "0 acme frames reached globex-only tail (filter works)" else fail "got $acme_frames acme frames on globex-only tail (expected 0)" fi if [[ $globex_frames -ge 1 ]]; then pass "$globex_frames globex frame(s) reached the tail" else fail "got $globex_frames globex frames (expected ≥1)" fi # ── Summary ───────────────────────────────────────────────── echo "" if [[ $fails -eq 0 ]]; then echo "🟢 M5 smoke PASS — all checks green" exit 0 else echo "🔴 M5 smoke FAIL — $fails check(s) failed" exit $fails fi