main.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778
  1. // Command tailcount subscribes to the M5 live tail
  2. // (GET /v1/tail/ws?token=…&company_id=…) and counts how many
  3. // events arrive within a timeout. Optionally filters to a
  4. // specific dedupe_key. Used by the M6.5 smoke (Step 3) to
  5. // prove the tail sees the storm even though the recipient
  6. // only gets 1 message.
  7. //
  8. // Usage:
  9. //
  10. // tailcount -url ws://localhost:8800/v1/tail/ws \
  11. // -token tail-dev-token-please-change-in-prod \
  12. // -company acme-001 \
  13. // -filter "m65-step3-1234" \
  14. // -timeout 15s
  15. //
  16. // On exit (timeout or signal), prints the count to stdout
  17. // in the form `count=<N>` so the smoke script can grep.
  18. package main
  19. import (
  20. "flag"
  21. "fmt"
  22. "log"
  23. "net/url"
  24. "strings"
  25. "time"
  26. "github.com/gorilla/websocket"
  27. )
  28. func main() {
  29. urlFlag := flag.String("url", "ws://localhost:8800/v1/tail/ws", "tail endpoint")
  30. token := flag.String("token", "", "tail token (BA_INGESTD_TAIL_TOKEN)")
  31. company := flag.String("company", "", "company_id filter (required)")
  32. filter := flag.String("filter", "", "sub-string filter: only count events containing this string")
  33. timeout := flag.Duration("timeout", 15*time.Second, "max wall time to listen")
  34. flag.Parse()
  35. if *token == "" || *company == "" {
  36. log.Fatal("token and company are required")
  37. }
  38. u, err := url.Parse(*urlFlag)
  39. if err != nil {
  40. log.Fatalf("parse url: %v", err)
  41. }
  42. q := u.Query()
  43. q.Set("token", *token)
  44. q.Set("company_id", *company)
  45. u.RawQuery = q.Encode()
  46. dialer := *websocket.DefaultDialer
  47. dialer.HandshakeTimeout = 5 * time.Second
  48. conn, resp, err := dialer.Dial(u.String(), nil)
  49. if err != nil {
  50. st := "<nil>"
  51. if resp != nil {
  52. st = resp.Status
  53. }
  54. log.Fatalf("dial: %v (status=%s)", err, st)
  55. }
  56. defer conn.Close()
  57. count := 0
  58. deadline := time.Now().Add(*timeout)
  59. conn.SetReadDeadline(time.Now().Add(*timeout + 1*time.Second))
  60. for time.Now().Before(deadline) {
  61. _, msg, err := conn.ReadMessage()
  62. if err != nil {
  63. break
  64. }
  65. s := string(msg)
  66. if *filter == "" || strings.Contains(s, *filter) {
  67. count++
  68. }
  69. }
  70. fmt.Printf("count=%d\n", count)
  71. }