From 2237fbda552130f279e10783b263e89b523ded4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Sat, 19 Sep 2026 20:06:36 +0200 Subject: [PATCH] more --- README.md | 6 + .../__pycache__/server.cpython-314.pyc | Bin 0 -> 15054 bytes cmd/needle-sidecar/eval/golden.jsonl | 40 ++++++ cmd/needle-sidecar/eval/run_eval.py | 76 ++++++++++ cmd/needle-sidecar/server.py | 52 ++++++- cmd/normalized-backfill/main.go | 13 +- cmd/server/main.go | 1 + internal/collector/handlers.go | 35 +++++ internal/normalization/normalizer.go | 41 +++++- internal/normalization/normalizer_test.go | 26 ++++ internal/storage/clickhouse_endpoints.go | 94 +++++++++++++ internal/storage/store.go | 26 ++++ otel-ui/src/components/TraceVisualization.tsx | 130 ++++++++++++++++-- .../contexts/ObservabilityContextProvider.tsx | 57 ++++---- otel-ui/src/lib/ingest.ts | 49 ++++++- 15 files changed, 594 insertions(+), 52 deletions(-) create mode 100644 cmd/needle-sidecar/__pycache__/server.cpython-314.pyc create mode 100644 cmd/needle-sidecar/eval/golden.jsonl create mode 100755 cmd/needle-sidecar/eval/run_eval.py create mode 100644 internal/storage/clickhouse_endpoints.go diff --git a/README.md b/README.md index 9c5b102..31ab8fe 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,12 @@ explanatory JSON body and the UI hides its "Trends" tab. for release-over-release comparison. Entries need a `service.release` attribute to appear here. - `GET /api/stats/services` - Distinct services seen in the rollup. +- `GET /api/stats/endpoints?hours=72&host=&service=&limit=100` - HTTP span + traffic grouped by normalized service, host, method, path, route, status, + user-agent, and safe browser request metadata. `spanCount` exposes nested + proxy/instrumentation duplication while `traceCount` approximates logical + requests. Origin and referrer are returned as host-only values; query + strings are stripped from paths. All accept `hours` (1..2160, default 72) and an optional `service` filter. diff --git a/cmd/needle-sidecar/__pycache__/server.cpython-314.pyc b/cmd/needle-sidecar/__pycache__/server.cpython-314.pyc new file mode 100644 index 0000000000000000000000000000000000000000..898c190ca7d2da72d0100dc2035c128e9f9d130a GIT binary patch literal 15054 zcmeG@TW}lKb-TdgMSvtog74Qtlt@t`2~wgY>OoNwMe0q_D@d{=PzVSt$(29=-d*Sc z#~G&W#8jG$D95ds#)+9m8Pij{!%m$UeYmYMbvyp>OqT*JG8=X5PBPQvqcgOnPBLyk zdd^)e9yBRgiSv<}UJ3W!z4zSbx#!$-&OO{*ke^3E_^*Hc+e^{)6!i=I&;zM~k4J+H zMNLo~B~uqEjy^)mG)WDzfuu&+NK!^-NNSQzBsI%sNDW6Uhpn>puuZnrYwh#oydk@r z8lnd)^r158V09+1=QtzBbQ-j;PFhRlf@f%N;>>z`xo{|Nkkxz6q@1Ok;;iLV*x8vX zl#AAYv;kyZ4rK8fkamD{J>>v8RJ^)sH|4P2vs{`6ku8V@Dca!o)X8!u8GgIqUM4|~?E z*Vj`+%-}wKH#4c5N}o?R;OJncM$Z9d6%7>+l?=ItN{7mZ%7-f2tlXv@I^8soaW+4# z_aS$SPXEv`*OE_hty_$l*_Bc-+cww{&ycOtsEsYJ=i19DdBad6*U|MnMTMY;8fs`m zX3q7%KN`rdc}wDZc>_Ej3iIq%cK;F9%ZK@KUKX#jUQq~)^6V)g#0LW+dx{sY2tht? ze=x|$WQjcqkgO01LUlxDrTF-GKooe14Mak0T!Kt*ASlPB*52rNEH3k&2pME6D@WO2 zI3P*F$aMnCUz0_EWZ@@7BY|OsL=vv$jTctd5^JyYG!_o>`}Ey+Fq%aabA*O z6IEke48KXMMoAW%Y1KRu6~_azgcDe&+7Ox-+hW&MhusTKOY@q=Kn)fO*OaKlo-Il|cq>51KP@#xoShoS`|> z8H0y`X`2(h93P8{GGImkY_>{(RQ71#O5jvb6k<>;2Ba}IDzdyNLL82cj`AUP_&Pf* zMz6y3S?=_SHr1@5P%YGY@n4esq@jp&5N#OYS$>mhHm)d=aK?>|jbZ_$YMn?X>lvqr9xzCBR=C)*cPi%X)S+@7@nIBD2v5J01uNK_MEK*emfc%swzI zga!FJi;Iy6rmw}qQGuZTDm0SX#9~}Xfy4*nB0%~B*p*lgg(AuiVkJh17Qk+9X)bmv zNkf0mu>P1B9p**ECw3#%LX<{q`Vc8(85#Sju8 z8{IBRab8Y(?yJzeUKc zTbpH?_yN`oKbR%+2x%ghsS?&;O7tF&jP^*o$a0MKvZV?qj9(!MeV|pSiluK zF2F??8D(Q|Nf5FdMUsn-j|*`7X%uGxZtI6G#EmeP*o+ZDk0Dba;oNlYS%L7We*GdZ zMdM!fix111}{Sv zP#BB?_b0Lo8;eH6tSpT4(Rdnt1$rlb(pvH7plBxX2t@D@|R9E4tbW zfOsgf=}alp*yW*nr{6!MoL-a8{xDe02fA*Jr6sU&MZLkJ$J4V?EO_$~(v?PVloNjhP<5t{>tlKZeHa+im zE9Zxm&?UuwSz#`Jo=6#J_xrL}o{BI$c`4}&rP9*`VW;Uk1W~8qazjL&P9f^q z80Wq|5JAEq0J9PgQZdgC2ZC^xLhQC{*D`WwJTgWE{t$4^h!8}jA{uFhhT#;~EaX$& z_<glmd!oO)9q5@>g~IsDK?i-*h$MKn$UUn-ne4-Dvb9Jh!~JF ziHddosmSTVS!$v&B4>agXCOsQ1HdE^IRn2Fawhv>hjC-rt3W|4kP!<=0`(8Hncx`Q zV3BA9NN0r9M4=kkBcO5=$cvYD!PN)C6c20@gv=lx1}uXx1)p$Bf$S_p?&Qf6E$kQ{ z2+LzEbg1*#K;*hShWBJV5)AVJQI}%ylcw;v^@Y&I}kGa zYScuQ=6h!q`w4|PL6DoMxZu0mwz+j^^FGKx;%mbK$RsR#_;Fw!dlS{Z;f+1Je4!UR zTDHFAYioWH;?&1*g4rX)_9nV~!x!3HcLrKV_P6#Az1a0qE40Wi>3pdb+hrGZYDbRal70*Rkiy-W;vz%iu36BZIDwmx3G@pMELyBRX`!%t)7iLP7K@TnUi8 z1`Gz047mX;2_}dMAM`?Ig374pPO2Xi;HyCDV(}Sz5Mfq?v#f3#ic!aa$Za6Qf}TvZ z;2cr`L`-_ufbc75j+P0#pX)ogzt?-@8BfF1-;)gWFI4?{-d`YfyIwXpIK zsL0|`s3EIsgvQAynm%cNncu;}KM8&{@+$wKy<%qK?5RsXTQaLV-U-3n7iR;oHv z#*NFZpbOHXYDSrlml9i!M~Sh7Ohwz3ul7Z%9&-`8A{IMEU=|kHD4PxWyB#TPalba3~aackq8-72nO|` zbqGi~7*5At{aP?!_joD{!8mr*8Smd^lIx#Q5Y2>D!nOD zvugbb9;Awplqf#b=WSsJ!T1E+ZIB$5HCh}7u@V&Rga-#gD~$JgjBjE0XNxB-?DMTB zWgaFmN+3|o8UXa!B~-zHaY7GZXJRBiJVFc&DU(W`p8$ZWD5(`Wo86YF3BXQD$OR{; zC97l6S}|{}P%5|IH7;29WEJdCcAi+U4&;_xdSJ~GOLWvAnGxiTkzQ9v-o+0x=1hRV zHY-AB>YJbzNqj}f)T3Fm+3@QR0AQfmsOFPZ!}St_dxjU!La}Pk?35&)hLW6p3Pt@X zEUnz9Ow9;3(rA}#Le;#rYI^H}wI0517TzdauvVrw=Yh5A^IQr|LS$S17tS&{(@?N7 za1`2ZM?u4^!4ti*8ooK2Sa;t{L9DWI7vIm>+9z%ujDhy76#=b<8HxZHG~=xZS=$fv z!?KZP&dtXZNA`iv%Cq`AE>`=b-_@(MEC3leEVaqhTa8-K_|NV9;F| zd@j9Y5KWvbf}1)+>thU}ZD~c|^7UFh1>1OG1Mc`L`^gn#pa9}&`}yR(f>D}_5*Cjp z?j(vmftZl8l6y20;%AO*R`8iK5Q~A;PSaX0fnueaH8|A}y{y{6Oe%y?tM;lU!ty^W zTb<>rreFYsE8r4Bf)@v1M?N#f<&oB|U%v|HaiG~M;x>z@hKLOqZN{hrBa|)0Js9nW zs5xJ=L0rHBBreg55#GA0*^hIRRAU$wLBBSl>FS68TEFQkSP)@+vIbKqJ|2VXi;tke zwAmn@!R8DSttPkWw1*x_L+4w6x%8fUlb;Z=76k)ROX=c}LCkMWwZ8!Lcu0u;8fs*jX{Xe#u!q z?Ok$~E0rB2?s~0XsqC8Rf2;Q$^WPksuk3=-Pbphf@v^ORjr5I2N*`~cXZZJ=!Bxi4SElzf zh~}OZL5GQ1eE8%*!fL$6o1SmX>b3GeCR_{+LL_jNyA8Vy*(c&64R7`75I4pRC^32{ zuN4HHDy|v@o|p}=wu(rTEw%x~9}(WYw)9@Ff{!(Ep4rEmFSd_99c|Pg+FVzJjH5#= z&xaxl34^=!Tqf1Ksi(5w(A2jgcuia_SGOZ&%8c2+==k2e9QTd-mMoDm=J48AJ82x- zmcDso`QCh;m4qMzHzyjjNbu%g%rQ>*_N6M)tr9UZJO)N{=FV=_QP%6HTaJ_rbMs9*-?%^BvM_Gn_&1--p&661v_6 zPZ8irfSC;cIHuQTQr`(Z?>m{!rCaFzu1tYv+cN!nEyzRupJ(7b0g~)0oRe$UcaYc$ zp2#_&hbP{KI0{SooI1A5ROl7NbBtLX1Ao-Fwb{#{-1_Wyb}ii6F{X#X-O_RbEl|6S zwUj_NZ~4sT(Ve{#1-deOoR5rxIe&@TFGc`VjAHZxMj~m6+Y|#Yu~-aDaG?E1g$S{Q!+%y2Yx9l8UD0F~yo z18Iuko0V$3%wHE@#7>Ry)TmlyG<=9S8D0`!#@WAwm5vcXlw@?sfYYIQc7PcMkAn@- zOgxOFVt{O9R|f6%zyg*IG$X5}pEH19F^(ktGbs0I3y4zD{h_`4VO8yP==Jb-!;5uo z^L1^Cr{`W#vaUB-b#U_7Qc3ykUAK1K?zz=7vpZSh0n>%SnqU6WX3wkg?ZhjIne_|h zO>^87vs_HMDsFe(>YDDKIgl)AS}bXuFKHz${LlAm_f@`M^b!f3< z{r!^lI@FHY&N<85-LoFWy=%d>TZj6BL-yV7d8J41Y8$w{@NZYo<}2>)3$7h{S0@%r zHr+4Tq(gPh9-OQFu%$=YyialOUvM2*DzCYnxRsdMkSyOg#lY&9oUTRZx_RfinXL=X zCiuEtc&l*1=_cjuypx@&Q#yJVoCnhd3(j>2mUk;}(TMzIzMClj(@U$sj&2Zq5Gk;Z0gGGf*JRR&W7(4ES0rSmnmiKGZnL) z^JVQ*Mlz=_WAeQ}s$X>Pn0N2EYe~BIEjahXq#rtpr@FufHbZ~ta3>u#zbva-ENi-7 z)|4!3o;>u>QFQazjbq4USSQ_Zj-SVvXeV%6sRRhyGlt#g(St9IP& z{IIfT^61B9mA6N4jZO=*#mTak$wMDCx4(1#p5=Y|U-CXUxVY`P``ezI-Z9hhgI&tm zbCbuHO5Ka48|F(l%#6(yBujfH`#*M--#&EfkW$k!JDhZNO!j@$zHQnx9r(WWb;oxd zGiMemTJCmF9s!5aqN?f6C1=faXbB8}jZ4n5X&YJreNgrqKhyihp&uTaJ$|<#xqff5 zt_N#Qy?*w)XJ;)m&+JG=9&&Si5z;cI(`sWbK{@ zj_#-S3}(mQJ{Am=tM)Nv&i@h*_}zkgFWd_%+m9+O$CB>ji|$kN?o)~vkNkpj=--`X z|FfJA1AZpq{rpoGd#0Uw&~88HFh1CQzrND1h!FpE;u;+ch(oMOvk8W6 z-OJ+pn<*xuhp&-b9U?0JDg*bnlydvoq_J>2am`D)gjeRW@vN#}KywKV9uw3rt@(=6 zJ+pr1g&#M~RL;7A8x_5MXm;n^!Mm=zfqy8!Yf}miJh1i>Gcoad7-+VuRzJB;{C+jh z?}w*=co^Kt(YgGEcp!|z2YeO!{p3ap3lTmN1#fd=zaN58MH!>FG5ULm)KWh&<%1V& zK!zK97@YYf=%C0C4%*;+fmbX#bNKzR5BXVz&6dGB>o+i!s{vgVS*q0RM8VlEjq?u zAC3mZ&=C~9#CS|r9RxyzgAkQ?A_8++zaoz?wXano&{##vVk$xEm;c|tu^!3ng`ap-#)b0Jcc7t z_YkO}pku+hh16EhTdNnXwaX@``AjN;5q{#_Z=rr_Y24po{AoK4F;WEh=0xC(HmMfB zKNJnZ2{uGQdXuh8M6eZKg@)j`!Hc&rW5<1L)0Ri}NW`Zj~M+aMyPrchjtNlb`dNDl(-e6ofx5sNjw0NY5{NG7(9xJhp^}! zh<|~w2Dx1|YivXOO9Voq(02GIm~&x-`l`lmta~)t-z)wblp!rjc*>WJG);fP7-{BF z9!2NBz|Ju~}%sEm9_oc!5a`o4%9~&xf=!U>|ur3ww!M%G)&n4yBIy3 z98|t@%78+_?SfkcN`1#%XVTe~bnKk4E!pf7=A}I6lzZy==@%56d&2yQnWD{#b=Ly5 zXUSw+G?m>ql_gCT69%A)oB7|&Ut-oRF}37>%@R|&#FQ^F)etWi*=h}sDVxnWVfw^H z`hfS%#oD&}wQb4T_PH$|)^3}Vl=b)x@8x| WI3^Pd#>$^DmT&Zd#}cFYI{Y7jhmVW^ literal 0 HcmV?d00001 diff --git a/cmd/needle-sidecar/eval/golden.jsonl b/cmd/needle-sidecar/eval/golden.jsonl new file mode 100644 index 0000000..d6ee971 --- /dev/null +++ b/cmd/needle-sidecar/eval/golden.jsonl @@ -0,0 +1,40 @@ +{"id": "ng-404-root", "src": "clickhouse", "expected": "log_404", "note": "", "text": "Service: nginx | Kind: span | Status: 404 | Route: /"} +{"id": "ng-401-repos", "src": "clickhouse", "expected": "auth_failure", "note": "polled ~700x/h by an unauthenticated client", "text": "Service: nginx | Kind: span | Status: 401 | Route: /api/repos/available"} +{"id": "ng-401-ctrl", "src": "clickhouse", "expected": "auth_failure", "note": "", "text": "Service: nginx | Kind: span | Status: 401 | Route: /api/controller/instances"} +{"id": "orch-401-repos", "src": "clickhouse", "expected": "auth_failure", "note": "same request as ng-401-repos, second span", "text": "Service: eventsourced-orchestrator | Kind: span | Title: GET / | Message: get / | Status: 401 | Route: /api/repos/available"} +{"id": "orch-401-chat", "src": "clickhouse", "expected": "auth_failure", "note": "", "text": "Service: eventsourced-orchestrator | Kind: span | Title: GET / | Message: get / | Status: 401 | Route: /api/sessions/{id}/collab/chat"} +{"id": "ng-401-manifest", "src": "clickhouse", "expected": "auth_failure", "note": "registry pull without token", "text": "Service: nginx | Kind: span | Status: 401 | Route: /v2/eventsourced-ai-worker/manifests/latest"} +{"id": "ng-503-store", "src": "clickhouse", "expected": "escalate_outage", "note": "upstream 503, real user-facing route", "text": "Service: nginx | Kind: span | Status: 503 | Route: /store/regions"} +{"id": "ng-500-root", "src": "clickhouse", "expected": "escalate_outage", "note": "", "text": "Service: nginx | Kind: span | Status: 500 | Route: /"} +{"id": "ng-500-wp", "src": "clickhouse", "expected": "security_alert", "note": "WordPress path that does not exist here", "text": "Service: nginx | Kind: span | Status: 500 | Route: /wp"} +{"id": "ng-500-blog", "src": "clickhouse", "expected": "security_alert", "note": "", "text": "Service: nginx | Kind: span | Status: 500 | Route: /blog"} +{"id": "ng-500-wpjson", "src": "clickhouse", "expected": "security_alert", "note": "", "text": "Service: nginx | Kind: span | Status: 500 | Route: /wp-json/batch/v1"} +{"id": "ng-500-wordpress", "src": "clickhouse", "expected": "security_alert", "note": "", "text": "Service: nginx | Kind: span | Status: 500 | Route: /wordpress"} +{"id": "ng-404-env", "src": "clickhouse", "expected": "security_alert", "note": "", "text": "Service: nginx | Kind: span | Status: 404 | Route: /.env"} +{"id": "ng-404-envlocal", "src": "clickhouse", "expected": "security_alert", "note": "", "text": "Service: nginx | Kind: span | Status: 404 | Route: /.env.local"} +{"id": "ng-404-userfiles", "src": "clickhouse", "expected": "security_alert", "note": "ambiguous: probe or stale link", "text": "Service: nginx | Kind: span | Status: 404 | Route: /userfiles"} +{"id": "ng-404-commit", "src": "clickhouse", "expected": "log_404", "note": "git UI link to missing commit", "text": "Service: nginx | Kind: span | Status: 404 | Route: /mats/go-cart-actor/commit/{id}"} +{"id": "ng-404-vite", "src": "clickhouse", "expected": "log_404", "note": "dev tooling request", "text": "Service: nginx | Kind: span | Status: 404 | Route: /__vite_rsc_findSourceMapURL"} +{"id": "ng-404-blob", "src": "clickhouse", "expected": "log_404", "note": "registry blob existence check", "text": "Service: nginx | Kind: span | Status: 404 | Route: /v2/eventsourced-ai/blobs/{id}"} +{"id": "ng-499-repos", "src": "clickhouse", "expected": "none", "note": "client closed request", "text": "Service: nginx | Kind: span | Status: 499 | Route: /api/repos/available"} +{"id": "ng-499-ctrl", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: nginx | Kind: span | Status: 499 | Route: /api/controller/instances"} +{"id": "ng-308-blog", "src": "clickhouse", "expected": "none", "note": "redirect", "text": "Service: nginx | Kind: span | Status: 308 | Route: /blog/"} +{"id": "ng-200-speed", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: nginx | Kind: span | Status: 200 | Route: /api/network-speed"} +{"id": "ng-200-matrix", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: nginx | Kind: span | Status: 200 | Route: /_matrix/federation/v1/send/{id}"} +{"id": "ng-200-overview", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: nginx | Kind: span | Status: 200 | Route: /api/overview"} +{"id": "sched-readyz", "src": "clickhouse", "expected": "none", "note": "health check", "text": "Service: scheduling | Kind: span | Title: GET /readyz | Message: get /readyz | Status: 200 | Route: /readyz"} +{"id": "cart-ping", "src": "clickhouse", "expected": "none", "note": "gRPC keepalive", "text": "Service: cart | Kind: span | Title: /control_plane_messages.ControlPlane/Ping | Message: /control_plane_messages.controlplane/ping"} +{"id": "rag-request", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: rust-rag | Kind: span | Title: request | Message: request"} +{"id": "orch-session-load", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: eventsourced-orchestrator | Kind: log | Message: session snapshot/replay loaded"} +{"id": "worker-dialerr", "src": "clickhouse", "expected": "none", "note": "dependency down; arguably escalate_outage - decide", "text": "Service: eventsourced-worker-code | Kind: span | Title: worker.terminal.connection | Message: *websocket.dialerror: websocket.dial ws://eventsourced-orchestrator.dev.svc.cluster.local:/...: connection refused"} +{"id": "worker-eof", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: eventsourced-worker-code | Kind: span | Title: worker.terminal.connection | Message: *errors.errorstring: eof"} +{"id": "ledger-skip", "src": "clickhouse", "expected": "none", "note": "", "text": "Service: ledger | Kind: log | Message: offsite replicate: content unchanged since last push - skipping"} +{"id": "web-cls", "src": "clickhouse", "expected": "record_vital", "note": "real frontend log; is this a vital or a console line?", "text": "Service: eventsourced-ai-web | Kind: log | Message: cls "} +{"id": "syn-js-1", "src": "synthetic", "expected": "client_js_crash", "note": "", "text": "Service: web-frontend | Kind: log | Title: TypeError: Cannot read properties of undefined | Message: Uncaught TypeError: Cannot read properties of undefined (reading 'map') | Route: /checkout"} +{"id": "syn-js-2", "src": "synthetic", "expected": "client_js_crash", "note": "", "text": "Service: shop-ui | Kind: log | Title: ChunkLoadError | Message: Loading chunk 42 failed | Route: /products/"} +{"id": "syn-vital-1", "src": "synthetic", "expected": "record_vital", "note": "", "text": "Service: web-frontend | Kind: metric | Title: web_vital TTFB | Message: TTFB 812ms"} +{"id": "syn-db-1", "src": "synthetic", "expected": "database_error", "note": "", "text": "Service: checkout | Kind: log | Title: database timeout after ms | Message: context deadline exceeded querying orders"} +{"id": "syn-db-2", "src": "synthetic", "expected": "database_error", "note": "", "text": "Service: inventory | Kind: log | Title: deadlock detected | Message: pq: deadlock detected while updating stock"} +{"id": "syn-db-3", "src": "synthetic", "expected": "database_error", "note": "", "text": "Service: orders | Kind: log | Title: connection pool exhausted | Message: too many connections in pool"} +{"id": "syn-oom", "src": "synthetic", "expected": "none", "note": "", "text": "Service: kubelet-proxy | Kind: log | Title: OOMKilled | Message: container exceeded memory limit and was restarted"} +{"id": "syn-retry", "src": "synthetic", "expected": "none", "note": "", "text": "Service: worker | Kind: span | Title: retry attempt 2 | Message: retrying after transient error"} diff --git a/cmd/needle-sidecar/eval/run_eval.py b/cmd/needle-sidecar/eval/run_eval.py new file mode 100755 index 0000000..93169ad --- /dev/null +++ b/cmd/needle-sidecar/eval/run_eval.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Evaluate the Needle sidecar against a reviewed golden set. + +Usage: run_eval.py [--url http://127.0.0.1:5005] [--golden eval/golden.jsonl] [--json out.json] + +Each golden row: {"id", "expected", "text"}; `expected` is a sidecar action name +or "none" when no tool should fit. The sidecar has no fallback tool, so "none" +rows measure how often it force-fits (see the confidence column). +""" +import argparse, json, os, sys, time, urllib.request +from collections import Counter, defaultdict + +HERE = os.path.dirname(os.path.abspath(__file__)) + + +def classify(url, text): + req = urllib.request.Request( + url.rstrip("/") + "/classify", + data=json.dumps({"text": text}).encode(), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=60) as r: + return json.load(r) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--url", default=os.environ.get("OTEL_NEEDLE_ADDR", "http://127.0.0.1:5005")) + ap.add_argument("--golden", default=os.path.join(HERE, "golden.jsonl")) + ap.add_argument("--json", help="write per-row results here") + ap.add_argument("--min-conf", type=float, default=0.0, + help="predictions below this confidence count as 'none'") + a = ap.parse_args() + + rows = [json.loads(l) for l in open(a.golden) if l.strip()] + results, confusion = [], Counter() + lat = [] + for row in rows: + try: + out = classify(a.url, row["text"]) + except Exception as e: + print(f"sidecar error on {row['id']}: {e}", file=sys.stderr) + sys.exit(2) + pred = out.get("action", "unclassified") + conf = out.get("confidence", 0.0) + if conf < a.min_conf or pred == "unclassified": + pred = "none" + lat.append(out.get("latency_ms", 0.0)) + confusion[(row["expected"], pred)] += 1 + results.append({**row, "predicted": pred, "confidence": conf, "ok": pred == row["expected"]}) + + labels = sorted({r["expected"] for r in results} | {r["predicted"] for r in results}) + print(f"{'label':<18}{'prec':>7}{'recall':>8}{'support':>9}") + for l in labels: + tp = confusion[(l, l)] + pp = sum(v for (e, p), v in confusion.items() if p == l) + sup = sum(v for (e, p), v in confusion.items() if e == l) + prec = tp / pp if pp else 0.0 + rec = tp / sup if sup else 0.0 + print(f"{l:<18}{prec:>7.2f}{rec:>8.2f}{sup:>9}") + + acc = sum(r["ok"] for r in results) / len(results) + print(f"\naccuracy {acc:.2%} over {len(results)} rows; " + f"median latency {sorted(lat)[len(lat)//2]:.0f}ms") + + wrong = [r for r in results if not r["ok"]] + if wrong: + print("\nmisses:") + for r in wrong: + print(f" {r['id']:<8} expected={r['expected']:<16} got={r['predicted']:<16} conf={r['confidence']:.2f}") + if a.json: + json.dump(results, open(a.json, "w"), indent=1) + + +if __name__ == "__main__": + main() diff --git a/cmd/needle-sidecar/server.py b/cmd/needle-sidecar/server.py index 0b61c3a..48e7d1c 100755 --- a/cmd/needle-sidecar/server.py +++ b/cmd/needle-sidecar/server.py @@ -8,6 +8,8 @@ import os import sys import json import time +import re +import hashlib from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler import needle @@ -50,6 +52,31 @@ def escalate_backend_outage(service: str, error_message: str): """Escalate severe backend 5xx crash or unhandled panic to on-call.""" return {"action": "escalate_outage", "service": service, "error": error_message} +@needle.tool +def report_unclassified(service: str, summary: str): + """Fallback when the event matches none of the other categories: retries, cancellations, restarts, OOM, health checks, or anything unclear.""" + return {"action": "unclassified", "service": service, "summary": summary} + +# --- Redaction: nothing PII-bearing should reach the model or the cache --- +_REDACTIONS = [ + (re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), ""), + (re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), ""), + (re.compile(r"\b[0-9a-fA-F]{8}-(?:[0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}\b"), ""), + (re.compile(r"\beyJ[\w-]+\.[\w-]+\.[\w-]+\b"), ""), + (re.compile(r"\b[0-9a-fA-F]{16,}\b"), ""), + (re.compile(r"([?&][\w.-]+=)[^&\s|]+"), r"\1"), +] + +def redact(text: str) -> str: + for rx, sub in _REDACTIONS: + text = rx.sub(sub, text) + return text[:1000] + +# Cache keyed on the redacted text: repeated signatures cost nothing. +CACHE_MAX = int(os.environ.get("NEEDLE_CACHE_MAX", 2000)) +_cache = {} +_stats = {"hits": 0, "misses": 0} + # Initialize Needle Agent once print(f"Loading Needle 3 model (Cactus-Compute/needle3)...") t0 = time.perf_counter() @@ -60,7 +87,8 @@ agent = needle.Needle(tools=[ log_missing_resource_404, report_database_error, report_auth_failure, - escalate_backend_outage + escalate_backend_outage, + report_unclassified ]) print(f"Needle 3 loaded in {(time.perf_counter() - t0):.2f}s") @@ -86,6 +114,7 @@ class NeedleHandler(BaseHTTPRequestHandler): "status": "ok", "available": True, "model": "Cactus-Compute/needle3", + "cache": {**_stats, "size": len(_cache)}, "tools": [ "client_js_crash", "security_alert", @@ -93,7 +122,8 @@ class NeedleHandler(BaseHTTPRequestHandler): "log_404", "database_error", "auth_failure", - "escalate_outage" + "escalate_outage", + "unclassified" ] } self.wfile.write(json.dumps(resp).encode("utf-8")) @@ -142,6 +172,20 @@ class NeedleHandler(BaseHTTPRequestHandler): self.wfile.write(json.dumps({"error": "Missing 'text' or telemetry fields"}).encode("utf-8")) return + text = redact(text) + key = hashlib.sha256(text.encode()).hexdigest() + cached = _cache.get(key) + if cached is not None: + _stats["hits"] += 1 + out = {**cached, "cached": True} + self.send_response(200) + self._send_cors() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps(out).encode("utf-8")) + return + _stats["misses"] += 1 + t_start = time.perf_counter() agent.reset() res = agent.run(text, strict=False) @@ -166,6 +210,10 @@ class NeedleHandler(BaseHTTPRequestHandler): "model": "Cactus-Compute/needle3" } + if len(_cache) >= CACHE_MAX: + _cache.pop(next(iter(_cache))) + _cache[key] = out + self.send_response(200) self._send_cors() self.send_header("Content-Type", "application/json") diff --git a/cmd/normalized-backfill/main.go b/cmd/normalized-backfill/main.go index 817b5eb..2445467 100644 --- a/cmd/normalized-backfill/main.go +++ b/cmd/normalized-backfill/main.go @@ -160,7 +160,9 @@ func createSchema(ctx context.Context, conn clickhouse.Conn, database string) er metric_value Float64, metric_count UInt64, metric_sum Float64, metric_min Float64, metric_max Float64, exception_type String, message_normalized String, stack_signature String, trace_id String, issue_fingerprint String, is_retry UInt8, is_timeout UInt8, is_cancellation UInt8, - exact_signature String, pattern_signature String, version UInt8 DEFAULT 1 + exact_signature String, pattern_signature String, version UInt8 DEFAULT 1, + method String DEFAULT '', host String DEFAULT '', user_agent String DEFAULT '', + origin_host String DEFAULT '', referer_host String DEFAULT '', fetch_site String DEFAULT '', fetch_mode String DEFAULT '' ) ENGINE = ReplacingMergeTree(version) PARTITION BY toYYYYMM(timestamp) ORDER BY (service, kind, pattern_signature, timestamp, event_id) @@ -175,6 +177,13 @@ func createSchema(ctx context.Context, conn clickhouse.Conn, database string) er fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_retry UInt8 DEFAULT 0", database), fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_timeout UInt8 DEFAULT 0", database), fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_cancellation UInt8 DEFAULT 0", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS method String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS host String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS user_agent String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS origin_host String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS referer_host String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS fetch_site String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS fetch_mode String DEFAULT ''", database), } { if err := conn.Exec(ctx, statement); err != nil { return err @@ -191,7 +200,7 @@ func appendEvent(batch driver.Batch, event normalization.Event) error { event.MetricValue, event.MetricCount, event.MetricSum, event.MetricMin, event.MetricMax, event.ExceptionType, event.MessageNormalized, event.StackSignature, event.TraceID, event.IssueFingerprint, boolUint8(event.IsRetry), boolUint8(event.IsTimeout), boolUint8(event.IsCancellation), event.ExactSignature, event.PatternSignature, - uint8(1), + uint8(1), event.Method, event.Host, event.UserAgent, event.OriginHost, event.RefererHost, event.FetchSite, event.FetchMode, ) } diff --git a/cmd/server/main.go b/cmd/server/main.go index 2aeefe4..7afafac 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -324,6 +324,7 @@ func main() { uiMux.Handle("/api/stats/errors", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsErrors))) uiMux.Handle("/api/stats/releases", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsReleases))) uiMux.Handle("/api/stats/services", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsServices))) + uiMux.Handle("/api/stats/endpoints", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleStatsEndpoints))) uiMux.HandleFunc("/ws", wsServer.HandleConnections) uiMux.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("./static")))) // Serve the agent instructions doc as a static route so an AI agent can diff --git a/internal/collector/handlers.go b/internal/collector/handlers.go index a8c9e42..f882856 100644 --- a/internal/collector/handlers.go +++ b/internal/collector/handlers.go @@ -852,6 +852,41 @@ func (c *Collector) HandleStatsServices(w http.ResponseWriter, r *http.Request) _ = json.NewEncoder(w).Encode(map[string]interface{}{"services": services}) } +// HandleStatsEndpoints returns normalized HTTP traffic groups. It is useful +// for diagnosing noisy hosts and proxy/instrumentation duplication without +// loading raw telemetry into the browser. +func (c *Collector) HandleStatsEndpoints(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if !c.statsEnabled() { + statsNotEnabled(w) + return + } + service, hours := parseStatsCommon(r) + host := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("host"))) + limit := parseLimit(r, 100) + if limit <= 0 { + limit = 100 + } + if limit > 500 { + limit = 500 + } + stats, err := c.statsQuerier.EndpointStats(r.Context(), host, service, hours, limit) + if err != nil { + log.Printf("failed to compute endpoint stats: %v", err) + http.Error(w, "failed to compute endpoint stats", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "hours": hours, + "host": host, + "endpoints": stats, + }) +} + // needleAddr returns the configured or default address for the Needle 3 sidecar. func needleAddr() string { addr := os.Getenv("OTEL_NEEDLE_ADDR") diff --git a/internal/normalization/normalizer.go b/internal/normalization/normalizer.go index e4c74a5..0f193fa 100644 --- a/internal/normalization/normalizer.go +++ b/internal/normalization/normalizer.go @@ -26,6 +26,13 @@ type Event struct { Service string `json:"service"` Release string `json:"release,omitempty"` Environment string `json:"environment,omitempty"` + Method string `json:"method,omitempty"` + Host string `json:"host,omitempty"` + UserAgent string `json:"userAgent,omitempty"` + OriginHost string `json:"originHost,omitempty"` + RefererHost string `json:"refererHost,omitempty"` + FetchSite string `json:"fetchSite,omitempty"` + FetchMode string `json:"fetchMode,omitempty"` Kind string `json:"kind"` Source string `json:"source"` IsError bool `json:"isError"` @@ -108,7 +115,7 @@ func normalizeTraces(ingest int64, envelope map[string]json.RawMessage) []Event if exceptionType != "" && message != "" { message = exceptionType + ": " + message } - isError := status == 2 || statusCode >= 400 || intAttr(attrs, "rpc.grpc.status_code") > 0 || exceptionType != "" + isError := status == 2 || isErrorStatus(statusCode) || intAttr(attrs, "rpc.grpc.status_code") > 0 || exceptionType != "" e := baseEvent(ingest, attrs, "span", item.TraceID) e.IsError, e.StatusCode, e.StatusClass = isError, statusCode, statusClass(statusCode) e.DurationMS = durationMS(item.Start, item.End) @@ -140,7 +147,7 @@ func normalizeLogs(ingest int64, envelope map[string]json.RawMessage) []Event { statusCode := intAttr(attrs, "http.response.status_code", "http.status_code") grpcCode := intAttr(attrs, "rpc.grpc.status_code") e := baseEvent(timestampOr(ingest, item.Timestamp), attrs, "log", item.TraceID) - e.IsError = item.SeverityNumber >= 17 || strings.EqualFold(item.SeverityText, "ERROR") || strings.EqualFold(item.SeverityText, "FATAL") || exceptionType != "" || statusCode >= 400 || grpcCode > 0 + e.IsError = item.SeverityNumber >= 17 || strings.EqualFold(item.SeverityText, "ERROR") || strings.EqualFold(item.SeverityText, "FATAL") || exceptionType != "" || isErrorStatus(statusCode) || grpcCode > 0 e.StatusCode, e.StatusClass = statusCode, statusClass(statusCode) e.Route, e.ExceptionType, e.MessageNormalized, e.StackSignature = normalizeHTTPRoute(attrs), exceptionType, normalizeMessage(message), firstFrame(stack) e.IsRetry, e.IsTimeout, e.IsCancellation = operationFlags(attrs, statusCode, grpcCode) @@ -301,7 +308,28 @@ func baseEvent(timestamp int64, attrs []attribute, kind, traceID string) Event { if source == "" { source = "backend" } - return Event{Timestamp: timestamp, NormalizerVersion: Version, Service: defaultString(stringAttr(attrs, "service.name"), "unknown"), Release: stringAttr(attrs, "service.release"), Environment: defaultString(stringAttr(attrs, "deployment.environment.name", "deployment.environment", "service.environment"), ""), Kind: kind, Source: source, TraceID: traceID} + ep := httpattr.Extract(func(keys ...string) string { return stringAttr(attrs, keys...) }) + return Event{ + Timestamp: timestamp, NormalizerVersion: Version, + Service: defaultString(stringAttr(attrs, "service.name"), "unknown"), + Release: stringAttr(attrs, "service.release"), + Environment: defaultString(stringAttr(attrs, "deployment.environment.name", "deployment.environment", "service.environment"), ""), + Method: ep.Method, Host: ep.Host, + UserAgent: stringAttr(attrs, "http.user_agent", "user_agent.original"), + OriginHost: attrHost(stringAttr(attrs, "http.request.header.origin", "http.origin", "origin")), + RefererHost: attrHost(stringAttr(attrs, "http.request.header.referer", "http.referer")), + FetchSite: stringAttr(attrs, "http.request.header.sec_fetch_site"), + FetchMode: stringAttr(attrs, "http.request.header.sec_fetch_mode"), + Kind: kind, Source: source, TraceID: traceID, + } +} + +func attrHost(raw string) string { + if raw == "" { + return "" + } + host, _ := httpattr.SplitURL(raw) + return host } func finish(e Event) Event { if e.IsError { @@ -445,3 +473,10 @@ func exception(events []spanEvent) (string, string, string) { } return "", "", "" } + +// isErrorStatus reports whether an HTTP status alone marks an event as an +// error. 404/410 are excluded: missing resources and scanner probes would +// otherwise dominate the error rate the spike label is built from. +func isErrorStatus(code int) bool { + return code >= 400 && code != 404 && code != 410 +} diff --git a/internal/normalization/normalizer_test.go b/internal/normalization/normalizer_test.go index c033f72..fc08836 100644 --- a/internal/normalization/normalizer_test.go +++ b/internal/normalization/normalizer_test.go @@ -106,3 +106,29 @@ func TestNormalizeHTTPRoutePrefersSpecificPathOverRootMount(t *testing.T) { t.Fatalf("expected specific URL path, got %q", events[0].Route) } } + +func TestNormalizeHTTPRequestMetadataIsSafeAndNormalized(t *testing.T) { + data := []byte(`{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"preview"}}]},"scopeSpans":[{"spans":[{"traceId":"trace-1","name":"GET /","attributes":[{"key":"http.method","value":{"stringValue":"GET"}},{"key":"http.host","value":{"stringValue":"preview.example.test:443"}},{"key":"http.target","value":{"stringValue":"/"}},{"key":"http.user_agent","value":{"stringValue":"Chrome"}},{"key":"http.request.header.origin","value":{"stringValue":"https://app.example.test"}},{"key":"http.request.header.referer","value":{"stringValue":"https://app.example.test/dashboard?secret=1"}},{"key":"http.request.header.sec_fetch_site","value":{"stringValue":"same-origin"}},{"key":"http.request.header.sec_fetch_mode","value":{"stringValue":"navigate"}}]}]}]}]}`) + events, err := NormalizeEntry("traces", 99, data) + if err != nil || len(events) != 1 { + t.Fatalf("normalize failed: %v %+v", err, events) + } + event := events[0] + if event.Method != "GET" || event.Host != "preview.example.test" || event.UserAgent != "Chrome" { + t.Fatalf("unexpected request metadata: %+v", event) + } + if event.OriginHost != "app.example.test" || event.RefererHost != "app.example.test" { + t.Fatalf("request metadata must retain hosts without paths: %+v", event) + } + if event.FetchSite != "same-origin" || event.FetchMode != "navigate" { + t.Fatalf("missing fetch metadata: %+v", event) + } +} + +func TestIsErrorStatusIgnoresMissingResources(t *testing.T) { + for code, want := range map[int]bool{200: false, 404: false, 410: false, 401: true, 500: true} { + if got := isErrorStatus(code); got != want { + t.Errorf("isErrorStatus(%d) = %v, want %v", code, got, want) + } + } +} diff --git a/internal/storage/clickhouse_endpoints.go b/internal/storage/clickhouse_endpoints.go new file mode 100644 index 0000000..8944264 --- /dev/null +++ b/internal/storage/clickhouse_endpoints.go @@ -0,0 +1,94 @@ +package storage + +import ( + "context" + "fmt" + "time" +) + +// EndpointStats reads raw trace spans because the production entries table is +// the source of truth and may not have the optional normalized_events table. +// The query intentionally returns both span and trace counts: a reverse proxy +// can legitimately produce nested spans for one request, as with nginx +// ingress -> upstream instrumentation. +func (s *ClickHouseStore) EndpointStats(ctx context.Context, host, service string, hours, limit int) ([]EndpointStat, error) { + if hours <= 0 { + hours = 72 + } + if limit <= 0 || limit > 500 { + limit = 100 + } + + where := "host != ''" + rawWhere := "type = 'traces' AND timestamp >= ?" + args := []interface{}{time.Now().UTC().Add(-time.Duration(hours) * time.Hour)} + if host != "" { + where += " AND host = ?" + args = append(args, host) + rawWhere += " AND positionCaseInsensitive(data, ?) > 0" + args = append(args, host) + } + if service != "" { + where += " AND service = ?" + args = append(args, service) + } + + query := fmt.Sprintf(` + SELECT service, host, method, path, if(route_raw = '/' AND path != '/', '', route_raw) AS route, status_code, user_agent, + origin_host, referer_host, fetch_site, fetch_mode, + count() AS span_count, uniqExact(trace_id) AS trace_count, + min(timestamp) AS first_seen, max(timestamp) AS last_seen + FROM ( + WITH JSONExtractArrayRaw(span, 'attributes') AS attrs + SELECT + timestamp, + JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'service.name', JSONExtractArrayRaw(JSONExtractRaw(resource_span, 'resource'), 'attributes')), 'value', 'stringValue') AS service, + JSONExtractString(span, 'traceId') AS trace_id, + upper(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.request.method', 'http.method'), attrs), 'value', 'stringValue'), ''), '')) AS method, + replaceRegexpOne(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'url.path', attrs), 'value', 'stringValue'), ''), JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.target', attrs), 'value', 'stringValue')), '[?#].*$', '') AS path, + JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.route', attrs), 'value', 'stringValue') AS route_raw, + lower(replaceRegexpOne(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.host', 'server.address', 'net.host.name'), attrs), 'value', 'stringValue'), ''), ''), ':\\d+$', '')) AS host, + toUInt16OrZero(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.response.status_code', 'http.status_code'), attrs), 'value', 'intValue'), ''), JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.response.status_code', 'http.status_code'), attrs), 'value', 'stringValue'))) AS status_code, + JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.user_agent', 'user_agent.original'), attrs), 'value', 'stringValue') AS user_agent, + replaceRegexpOne(replaceRegexpOne(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.request.header.origin', 'http.origin', 'origin'), attrs), 'value', 'stringValue'), '^https?://', ''), '/.*$', '') AS origin_host, + replaceRegexpOne(replaceRegexpOne(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.request.header.referer', 'http.referer'), attrs), 'value', 'stringValue'), '^https?://', ''), '/.*$', '') AS referer_host, + JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.request.header.sec_fetch_site', attrs), 'value', 'stringValue') AS fetch_site, + JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.request.header.sec_fetch_mode', attrs), 'value', 'stringValue') AS fetch_mode + FROM ( + SELECT timestamp, span + FROM %s.entries + ARRAY JOIN JSONExtractArrayRaw(data, 'resourceSpans') AS resource_span + ARRAY JOIN JSONExtractArrayRaw(resource_span, 'scopeSpans') AS scope_span + ARRAY JOIN JSONExtractArrayRaw(scope_span, 'spans') AS span + WHERE %s + ) + ) AS spans + WHERE %s + GROUP BY service, host, method, path, route_raw, status_code, user_agent, origin_host, referer_host, fetch_site, fetch_mode + ORDER BY trace_count DESC, span_count DESC, last_seen DESC + LIMIT ?`, s.db, rawWhere, where) + args = append(args, limit) + + rows, err := s.conn.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("clickhouse endpoint stats: %w", err) + } + defer rows.Close() + + var out []EndpointStat + for rows.Next() { + var stat EndpointStat + if err := rows.Scan( + &stat.Service, &stat.Host, &stat.Method, &stat.Path, &stat.Route, &stat.StatusCode, + &stat.UserAgent, &stat.OriginHost, &stat.RefererHost, &stat.FetchSite, + &stat.FetchMode, &stat.SpanCount, &stat.TraceCount, &stat.FirstSeen, &stat.LastSeen, + ); err != nil { + return nil, err + } + if stat.Route == "/" && stat.Path != "/" { + stat.Route = "" + } + out = append(out, stat) + } + return out, rows.Err() +} diff --git a/internal/storage/store.go b/internal/storage/store.go index 7826dfe..cf75519 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "git.tornberg.me/go-otel/internal/websocket" + "time" ) // SearchOptions narrows a Search call beyond the free-text query. @@ -38,6 +39,31 @@ type StatsQuerier interface { ReleaseSeries(ctx context.Context, service string, hours int) (map[string][]StatsPoint, error) // Services lists distinct services seen in the rollup. Services(ctx context.Context) ([]string, error) + // EndpointStats groups HTTP spans by normalized request identity and safe + // browser metadata. SpanCount exposes proxy/instrumentation duplication; + // TraceCount is the better approximation of logical requests. + EndpointStats(ctx context.Context, host, service string, hours, limit int) ([]EndpointStat, error) +} + +// EndpointStat is a bounded, privacy-conscious HTTP traffic aggregate. Origin +// and referer are host-only values; paths and query strings are normalized by +// the query so secrets are not returned as part of endpoint diagnostics. +type EndpointStat struct { + Service string `json:"service"` + Host string `json:"host"` + Method string `json:"method"` + Path string `json:"path"` + Route string `json:"route,omitempty"` + StatusCode uint16 `json:"statusCode"` + UserAgent string `json:"userAgent,omitempty"` + OriginHost string `json:"originHost,omitempty"` + RefererHost string `json:"refererHost,omitempty"` + FetchSite string `json:"fetchSite,omitempty"` + FetchMode string `json:"fetchMode,omitempty"` + SpanCount uint64 `json:"spanCount"` + TraceCount uint64 `json:"traceCount"` + FirstSeen time.Time `json:"firstSeen"` + LastSeen time.Time `json:"lastSeen"` } // TraceSummary is the per-trace rollup shown in overview tables: everything diff --git a/otel-ui/src/components/TraceVisualization.tsx b/otel-ui/src/components/TraceVisualization.tsx index e48da45..810adeb 100644 --- a/otel-ui/src/components/TraceVisualization.tsx +++ b/otel-ui/src/components/TraceVisualization.tsx @@ -21,6 +21,7 @@ import { import type { TraceData, Span } from "../lib/types"; import { formatDuration, formatRelativeTime } from "../lib/timeUtils"; import { formatAttributeValue, getServicePalette, getSpanColor } from "../lib/otel-utils"; +import { extractEndpoint } from "../lib/http-endpoint"; import { useLogsForTraceId } from "../hooks/useLogsForTraceId"; import { useObservabilityData } from "../hooks/useObservabilityData"; import { LogEntryView } from "./Tabs"; @@ -47,6 +48,8 @@ export type ProcessedTrace = { services: string[]; hasError: boolean; errorCount: number; + /** Normalized HTTP operation used for display/grouping; empty for non-HTTP traces. */ + endpoint: string; }; // Flattened item for virtualized tree rendering @@ -129,20 +132,26 @@ const flattenSpanTree = ( return result; }; +const getSpanEndpointLabel = (span: ProcessedSpan): string => { + const endpoint = extractEndpoint((...keys) => { + for (const key of keys) { + const value = span.formattedAttributes[key]; + if (value) return value; + } + return ""; + }); + const path = endpoint.route || endpoint.path; + return path ? `${endpoint.method || "HTTP"} ${path}` : ""; +}; + const getOperationDisplayName = (span: ProcessedSpan) => { let name = span.name; if (!name || name === "/" || name === "GET /") { - const method = span.formattedAttributes["http.method"]; - const target = span.formattedAttributes["http.target"]; - const route = span.formattedAttributes["http.route"]; - const urlPath = span.formattedAttributes["url.path"]; + const endpointLabel = getSpanEndpointLabel(span); const rpcMethod = span.formattedAttributes["rpc.method"]; - if (route) name = route; - else if (method && target) name = `${method} ${target}`; + if (endpointLabel) name = endpointLabel; else if (rpcMethod) name = rpcMethod; - else if (urlPath) name = urlPath; - else if (method) name = method; } return name || "span"; }; @@ -1211,6 +1220,9 @@ const TraceDirectoryTable = ({ onSortByChange, allServices, onSwitchToWaterfall, + groupByEndpoint, + groupCounts, + onGroupByEndpoint, }: { traces: ProcessedTrace[]; activeTraceId: string | null; @@ -1225,8 +1237,13 @@ const TraceDirectoryTable = ({ onSortByChange: (s: any) => void; allServices: string[]; onSwitchToWaterfall: () => void; + groupByEndpoint: boolean; + groupCounts: Record; + onGroupByEndpoint: (grouped: boolean) => void; }) => { const [copiedId, setCopiedId] = useState(null); + const [page, setPage] = useState(0); + const pageSize = 50; const copyTraceId = (id: string, e: React.MouseEvent) => { e.stopPropagation(); @@ -1243,6 +1260,17 @@ const TraceDirectoryTable = ({ return traces.filter((t) => t.hasError).length; }, [traces]); + const pageCount = Math.max(1, Math.ceil(traces.length / pageSize)); + const visibleTraces = traces.slice(page * pageSize, (page + 1) * pageSize); + + useEffect(() => { + setPage(0); + }, [searchQuery, filter, serviceFilter, sortBy, groupByEndpoint]); + + useEffect(() => { + setPage((current) => Math.min(current, pageCount - 1)); + }, [pageCount]); + return (
{/* Directory Header & Toolbar */} @@ -1263,7 +1291,7 @@ const TraceDirectoryTable = ({ )}
-

+

Browse, search, and inspect distributed trace execution flows

@@ -1362,6 +1390,18 @@ const TraceDirectoryTable = ({ )} + + {/* Sort Dropdown */}
@@ -1411,7 +1451,7 @@ const TraceDirectoryTable = ({ - {traces.map((trace) => { + {visibleTraces.map((trace) => { const isSelected = trace.id === activeTraceId; const rootService = trace.spans[0]?.service || trace.services[0] || "unknown"; const palette = getServicePalette(rootService); @@ -1455,8 +1495,13 @@ const TraceDirectoryTable = ({ {trace.name} - root: {trace.spans[0]?.name || trace.name} + {trace.endpoint ? `endpoint: ${trace.endpoint}` : `root: ${trace.spans[0]?.name || trace.name}`} + {groupCounts[trace.endpoint] > 1 && ( + + {groupCounts[trace.endpoint]} traces in this endpoint group + + )}
@@ -1548,6 +1593,33 @@ const TraceDirectoryTable = ({ )} + + {traces.length > pageSize && ( +
+ + {page * pageSize + 1}–{Math.min((page + 1) * pageSize, traces.length)} of {traces.length} + +
+ + Page {page + 1} / {pageCount} + +
+
+ )} ); }; @@ -1613,7 +1685,9 @@ export const processTraces = ( const traceList: ProcessedTrace[] = Object.entries(traceGroups).map(([id, spans]) => { const orderedSpans = spans.sort((a, b) => a.startTime - b.startTime); const displayName = - orderedSpans.find((span) => span.name)?.name || `Trace ${id.slice(0, 8)}`; + (orderedSpans.find((span) => span.name) && + getOperationDisplayName(orderedSpans.find((span) => span.name)!)) || + `Trace ${id.slice(0, 8)}`; const startTime = orderedSpans.reduce( (min, s) => Math.min(min, s.startTime), Number.POSITIVE_INFINITY, @@ -1625,6 +1699,9 @@ export const processTraces = ( const duration = Math.max(endTime - startTime, 0); const services = Array.from(new Set(orderedSpans.map((s) => s.service))); const errorCount = orderedSpans.filter((s) => s.hasError).length; + const endpointSpan = + orderedSpans.find((span) => span.kind === 2 && getSpanEndpointLabel(span)) ?? + orderedSpans.find((span) => getSpanEndpointLabel(span)); return { id, @@ -1636,6 +1713,7 @@ export const processTraces = ( services, hasError: errorCount > 0, errorCount, + endpoint: endpointSpan ? getSpanEndpointLabel(endpointSpan) : "", }; }); @@ -1665,6 +1743,7 @@ export const TraceVisualization = ({ traces }: TraceVisualizationProps) => { const [traceSortBy, setTraceSortBy] = useState< "recent" | "duration-desc" | "duration-asc" | "errors" | "spans" >("recent"); + const [groupByEndpoint, setGroupByEndpoint] = useState(false); const orderedTraces: ProcessedTrace[] = useMemo( () => processTraces(safeTraces, logCountBySpanId), @@ -1702,6 +1781,27 @@ export const TraceVisualization = ({ traces }: TraceVisualizationProps) => { }); }, [orderedTraces, traceFilter, traceServiceFilter, traceSearchQuery, traceSortBy]); + const groupCounts = useMemo(() => { + const counts: Record = {}; + filteredTraceList.forEach((trace) => { + if (trace.endpoint) counts[trace.endpoint] = (counts[trace.endpoint] ?? 0) + 1; + }); + return counts; + }, [filteredTraceList]); + + const directoryTraces = useMemo(() => { + if (!groupByEndpoint) return filteredTraceList; + const seen = new Set(); + return filteredTraceList.filter((trace) => { + // Non-HTTP traces stay individual; grouping them as "unknown" would be + // misleading and would hide otherwise unrelated trace types. + const key = trace.endpoint || `trace:${trace.id}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }); + }, [filteredTraceList, groupByEndpoint]); + // Set default active trace if none selected or if active trace is outside current filtered list useEffect(() => { if (filteredTraceList.length > 0) { @@ -1862,7 +1962,7 @@ export const TraceVisualization = ({ traces }: TraceVisualizationProps) => { {/* Main View: Directory Table or Waterfall */} {viewMode === "directory" ? ( setActiveTraceId(id)} searchQuery={traceSearchQuery} @@ -1875,6 +1975,9 @@ export const TraceVisualization = ({ traces }: TraceVisualizationProps) => { onSortByChange={setTraceSortBy} allServices={allServices} onSwitchToWaterfall={() => setViewMode("waterfall")} + groupByEndpoint={groupByEndpoint} + groupCounts={groupCounts} + onGroupByEndpoint={setGroupByEndpoint} /> ) : ( @@ -1884,4 +1987,3 @@ export const TraceVisualization = ({ traces }: TraceVisualizationProps) => { }; export default TraceVisualization; - diff --git a/otel-ui/src/contexts/ObservabilityContextProvider.tsx b/otel-ui/src/contexts/ObservabilityContextProvider.tsx index 4ca4671..6e6d0de 100644 --- a/otel-ui/src/contexts/ObservabilityContextProvider.tsx +++ b/otel-ui/src/contexts/ObservabilityContextProvider.tsx @@ -15,7 +15,7 @@ import type { } from "../lib/types" import { ObservabilityContext, type TimeRangeFilter } from "./ObservabilityContextValue" import { getServiceName, formatAttributeValue, isSpanError } from "../lib/otel-utils" -import { partitionMessages, mergeLogsByTraceId } from "../lib/ingest" +import { addFifoSet, appendFifo, partitionMessages, mergeLogsByTraceId, STREAM_LIMITS } from "../lib/ingest" import { authFetch, tokens } from "../lib/auth" interface ObservabilityProviderProps { @@ -257,21 +257,17 @@ export function ObservabilityProvider({ if (queued.length > 0) { const partitioned = partitionMessages(queued) if (partitioned.traces.length) { - setTraces((prev) => [...prev, ...partitioned.traces]) + setTraces((prev) => appendFifo(prev, partitioned.traces, STREAM_LIMITS.traces)) } if (partitioned.metrics.length) { - setMetrics((prev) => [...prev, ...partitioned.metrics]) + setMetrics((prev) => appendFifo(prev, partitioned.metrics, STREAM_LIMITS.metrics)) } if (partitioned.logs.length) { - setLogs((prev) => [...prev, ...partitioned.logs]) + setLogs((prev) => appendFifo(prev, partitioned.logs, STREAM_LIMITS.logs)) setLogsByTraceId((prev) => mergeLogsByTraceId(prev, partitioned.logsByTraceId)) } if (partitioned.traceIds.size) { - setKnownTraceIds((prev) => { - const next = new Set(prev) - partitioned.traceIds.forEach((id) => next.add(id)) - return next - }) + setKnownTraceIds((prev) => addFifoSet(prev, partitioned.traceIds, STREAM_LIMITS.knownTraceIds)) } } }, []) @@ -623,11 +619,11 @@ export function ObservabilityProvider({ partitioned.metrics.length, "metrics", ) - setTraces(partitioned.traces) - setMetrics(partitioned.metrics) - setLogs(partitioned.logs) - setLogsByTraceId(partitioned.logsByTraceId) - setKnownTraceIds(partitioned.traceIds) + setTraces(partitioned.traces.slice(-STREAM_LIMITS.traces)) + setMetrics(partitioned.metrics.slice(-STREAM_LIMITS.metrics)) + setLogs(partitioned.logs.slice(-STREAM_LIMITS.logs)) + setLogsByTraceId(mergeLogsByTraceId({}, partitioned.logsByTraceId)) + setKnownTraceIds(addFifoSet(new Set(), partitioned.traceIds, STREAM_LIMITS.knownTraceIds)) } catch (error) { console.error("Error fetching history:", error) } @@ -660,7 +656,11 @@ export function ObservabilityProvider({ // If stream is paused, buffer the incoming signal if (!isStreamingRef.current) { - bufferedMessagesRef.current.push(data as OtelMessage) + bufferedMessagesRef.current = appendFifo( + bufferedMessagesRef.current, + [data as OtelMessage], + STREAM_LIMITS.bufferedMessages, + ) setBufferedCount(bufferedMessagesRef.current.length) return } @@ -668,21 +668,17 @@ export function ObservabilityProvider({ const partitioned = partitionMessages([data as OtelMessage]) if (partitioned.traces.length) { - setTraces((prev) => [...prev, ...partitioned.traces]) + setTraces((prev) => appendFifo(prev, partitioned.traces, STREAM_LIMITS.traces)) } if (partitioned.metrics.length) { - setMetrics((prev) => [...prev, ...partitioned.metrics]) + setMetrics((prev) => appendFifo(prev, partitioned.metrics, STREAM_LIMITS.metrics)) } if (partitioned.logs.length) { - setLogs((prev) => [...prev, ...partitioned.logs]) + setLogs((prev) => appendFifo(prev, partitioned.logs, STREAM_LIMITS.logs)) setLogsByTraceId((prev) => mergeLogsByTraceId(prev, partitioned.logsByTraceId)) } if (partitioned.traceIds.size) { - setKnownTraceIds((prev) => { - const next = new Set(prev) - partitioned.traceIds.forEach((id) => next.add(id)) - return next - }) + setKnownTraceIds((prev) => addFifoSet(prev, partitioned.traceIds, STREAM_LIMITS.knownTraceIds)) } } catch (error) { console.error("Error parsing WebSocket message:", error) @@ -734,21 +730,22 @@ export function ObservabilityProvider({ const partitioned = partitionMessages(data ?? []) if (partitioned.traces.length) { - setTraces((prev) => [...prev, ...partitioned.traces]) + setTraces((prev) => appendFifo(prev, partitioned.traces, STREAM_LIMITS.traces)) } if (partitioned.metrics.length) { - setMetrics((prev) => [...prev, ...partitioned.metrics]) + setMetrics((prev) => appendFifo(prev, partitioned.metrics, STREAM_LIMITS.metrics)) } if (partitioned.logs.length) { - setLogs((prev) => [...prev, ...partitioned.logs]) + setLogs((prev) => appendFifo(prev, partitioned.logs, STREAM_LIMITS.logs)) setLogsByTraceId((prev) => mergeLogsByTraceId(prev, partitioned.logsByTraceId)) } setKnownTraceIds((prev) => { - const next = new Set(prev) - partitioned.traceIds.forEach((id) => next.add(id)) - next.add(activeTraceId) - return next + return addFifoSet( + prev, + new Set([...partitioned.traceIds, activeTraceId]), + STREAM_LIMITS.knownTraceIds, + ) }) }) .catch((error) => { diff --git a/otel-ui/src/lib/ingest.ts b/otel-ui/src/lib/ingest.ts index 6d3a2d2..23e65f0 100644 --- a/otel-ui/src/lib/ingest.ts +++ b/otel-ui/src/lib/ingest.ts @@ -5,6 +5,37 @@ import type { LogEntry, MetricData, OtelMessage, TraceData } from "./types" import { normalizeLogEntry, normalizeMetricData, normalizeTrace } from "./normalize" +/** + * Live telemetry is a rolling window. Without a bound, a long-lived browser + * session eventually retains every span/log/metric it has ever received. + */ +export const STREAM_LIMITS = { + traces: 500, + logs: 1_000, + metrics: 250, + bufferedMessages: 2_000, + knownTraceIds: 5_000, + logsPerTrace: 100, + logLinks: 10_000, +} as const + +/** Append items while retaining only the newest `limit` entries. */ +export function appendFifo(previous: T[], additions: T[], limit: number): T[] { + if (additions.length === 0) return previous + const combined = previous.length + additions.length <= limit + ? [...previous, ...additions] + : [...previous, ...additions].slice(-limit) + return combined +} + +/** Add ids in arrival order while retaining only the newest ids. */ +export function addFifoSet(previous: Set, additions: Iterable, limit: number): Set { + const next = new Set(previous) + for (const value of additions) next.add(value) + while (next.size > limit) next.delete(next.values().next().value as T) + return next +} + export interface PartitionedMessages { traces: TraceData[] metrics: MetricData[] @@ -75,11 +106,27 @@ export function partitionMessages(messages: OtelMessage[]): PartitionedMessages export function mergeLogsByTraceId( prev: Record, additions: Record, + limit = STREAM_LIMITS.logsPerTrace, + maxTraceIds = STREAM_LIMITS.knownTraceIds, + maxTotalEntries = STREAM_LIMITS.logLinks, ): Record { if (Object.keys(additions).length === 0) return prev const next = { ...prev } Object.entries(additions).forEach(([traceId, entries]) => { - next[traceId] = [...(next[traceId] ?? []), ...entries] + next[traceId] = appendFifo(next[traceId] ?? [], entries, limit) }) + const traceIds = Object.keys(next) + while (traceIds.length > maxTraceIds) { + delete next[traceIds.shift()!] + } + let totalEntries = Object.values(next).reduce((sum, entries) => sum + entries.length, 0) + while (totalEntries > maxTotalEntries) { + const oldestTraceId = Object.keys(next)[0] + if (!oldestTraceId) break + const entries = next[oldestTraceId] + entries.shift() + totalEntries-- + if (entries.length === 0) delete next[oldestTraceId] + } return next }