|
|
@@ -0,0 +1,395 @@
|
|
|
+#!/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
|
Luis Rosales