code wiki / _hdl_build / nx_apm_collect.nx
`nx_apm_collect.nx` source

↩ module page · 349 lines · 21145 B
1// nx_apm_collect.nx -- SOVEREIGN NAS-side INTERNAL-HTTP concurrency collector (the hardware-up APM layer).
2//
3// WHY (the gap nx_apm_sweep cannot fill): nx_apm_sweep probes nishifamily.com over the FULL public TLS
4// chain from the workstation. That is the right "what a browser sees" view, but each probe pays an ~800ms
5// Nishi TLS handshake whose cost is CPU-bound on this 21-workstream-contended workstation, so conc>1
6// latency is handshake-noise, not server truth -> it cannot cleanly measure a daemon's CONCURRENCY health.
7// This organ runs ON THE NAS (where the daemons live) and probes their plaintext loopback ports DIRECTLY
8// over HTTP/1.1 with "Connection: close" -- no TLS, no contention -> the latency it measures IS the daemon.
9//
10// It fires `conc` SIMULTANEOUS cold probes per target (= conc browser tabs), with a per-probe SO_RCVTIMEO
11// backstop so a truly-hung socket can never wedge the collector forever (rule #14, graceful degradation).
12// It distinguishes GREEN / AMBER(stall>2s) / RED(timeout|err) / DOWN(connection refused) -- the last is the
13// liar-kill: a daemon that is ABSENT must never read as healthy. Targets are DATA-DRIVEN (rule #11) from
14// knowledge/status/apm_targets.tsv; results append to a durable time-series knowledge/status/apm_collect.tsv.
15//
16// Single responsibility (rule #9): HTTP latency/concurrency only. Per-daemon CPU/mem is nx_perf_sentinel's
17// job already; the dashboard rung joins the two time-series. Proven by nx_apm_collect_gate (mock daemons).
18//
19// usage: nx_apm_collect            (reads knowledge/status/apm_targets.tsv)
20// license_tier: ORIGINAL
21import "nx_syscalls.nx"
22import "nx_connect.nx"   // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
23const AC_MAGIC_65536: i64 = 65536
24const AC_MAGIC_1024: i64 = 1024
25
26const AC_STALL_MS:  i64 = 2000
27const AC_TIMEOUT_S: i64 = 22                       // > the measured ~20s :8791 wedge, so we still capture it
28const AC_TARGETS:   *u8 = "knowledge/status/apm_targets.tsv"
29const AC_METRICS:   *u8 = "knowledge/status/apm_collect.tsv"
30
31// per-probe verdicts
32const AC_V_OK:      i64 = 0
33const AC_V_SOCK:    i64 = 1
34const AC_V_DOWN:    i64 = 2                         // connect refused -> daemon absent
35const AC_V_SEND:    i64 = 3
36const AC_V_TIMEOUT: i64 = 4                         // no bytes before SO_RCVTIMEO
37
38// per-target result codes (returned by ac_burst; the gate asserts on these)
39const AC_R_GREEN: i64 = 0
40const AC_R_AMBER: i64 = 1
41const AC_R_RED:   i64 = 2
42const AC_R_DOWN:  i64 = 3
43
44// ===== leaf helpers (decimal/string/io) ==========================
45func ac_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
46func ac_w(s: *u8) -> i64 { return sys_write(1, s, ac_slen(s)) }
47func ac_atoi(s: *u8) -> i64 {
48    var v: i64 = 0; var i: i64 = 0
49    while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v*10 + (c-48) } } i = i + 1 }
50    return v
51}
52// append decimal v into buf at off; returns new off
53func ac_n(buf: *u8, off: i64, v: i64) -> i64 {
54    var o: i64 = off; var m: i64 = v
55    if m < 0 { buf[o] = 45 as u8; o = o + 1; m = 0 - m }
56    let t: *u8 = sys_mmap(24); var k: i64 = 0
57    if m == 0 { t[0] = 48 as u8; k = 1 }
58    while m > 0 { t[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
59    var i: i64 = 0; while i < k { buf[o + i] = t[k - 1 - i]; i = i + 1 }
60    return o + k
61}
62// append C-string s into buf at off; returns new off
63func ac_s(buf: *u8, off: i64, s: *u8) -> i64 { var o: i64 = off; var i: i64 = 0; while s[i] != (0 as u8) { buf[o] = s[i]; o = o + 1; i = i + 1 } return o }
64func ac_wn(v: i64) -> i64 { let b: *u8 = sys_mmap(24); let e: i64 = ac_n(b, 0, v); return sys_write(1, b, e) }
65func ac_wpad(v: i64, w: i64) -> i64 {
66    let b: *u8 = sys_mmap(24); let e: i64 = ac_n(b, 0, v)
67    var pad: i64 = w - e; while pad > 0 { sys_write(1, " " as *u8, 1); pad = pad - 1 }
68    return sys_write(1, b, e)
69}
70func ac_sort(a: *i64, n: i64) -> i64 {
71    var i: i64 = 1
72    while i < n {
73        let kv: i64 = a[i]; var j: i64 = i - 1; var go: i64 = 1
74        while go == 1 { if j >= 0 { if a[j] > kv { a[j+1] = a[j]; j = j - 1 } else { go = 0 } } else { go = 0 } }
75        a[j+1] = kv; i = i + 1
76    }
77    return 0
78}
79// 16-byte sockaddr_in for a.b.c.d:port
80func ac_addr(out: *u8, a: i64, b: i64, c: i64, d: i64, port: i64) -> i64 {
81    out[0] = 2 as u8; out[1] = 0 as u8
82    out[2] = ((port >> 8) & 0xff) as u8; out[3] = (port & 0xff) as u8
83    out[4] = a as u8; out[5] = b as u8; out[6] = c as u8; out[7] = d as u8
84    var i: i64 = 8; while i < 16 { out[i] = 0 as u8; i = i + 1 }
85    return 16
86}
87// "GET <path> HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
88func ac_build_get(req: *u8, path: *u8) -> i64 {
89    var p: i64 = ac_s(req, 0, "GET " as *u8)
90    p = ac_s(req, p, path)
91    p = ac_s(req, p, " HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" as *u8)
92    return p
93}
94// Small bounded file read (NOT sys_read_file -- that mmaps 4 GiB/call and would balloon parent VA).
95func ac_read_small(path: *u8, buf: *u8, cap: i64) -> i64 {
96    let fd: i64 = sys_openat_rd(path)
97    if fd < 0 { return 0 - 1 }
98    var off: i64 = 0; var go: i64 = 1
99    while go == 1 {
100        if off >= cap { go = 0 }
101        else { let r: i64 = sys_read(fd, buf + off, cap - off); if r <= 0 { go = 0 } else { off = off + r } }
102    }
103    sys_close(fd)
104    return off
105}
106// copy one tab/newline-delimited field [pos..) into out; stop[0]=0 tab,1 newline,2 eof; returns next pos.
107func ac_field(buf: *u8, n: i64, pos: i64, out: *u8, outcap: i64, stop: *i64) -> i64 {
108    var i: i64 = pos; var w: i64 = 0; var done: i64 = 0
109    while done == 0 {
110        if i >= n { stop[0] = 2; done = 1 }
111        else { let c: i64 = buf[i] as i64
112               if c == 9 { stop[0] = 0; i = i + 1; done = 1 }
113               else { if c == 10 { stop[0] = 1; i = i + 1; done = 1 }
114               else { if w < outcap - 1 { out[w] = buf[i]; w = w + 1 } i = i + 1 } } }
115    }
116    out[w] = 0 as u8
117    return i
118}
119
120// ===== one internal probe ========================================
121// connect 127.0.0.1:port, GET path, drain to close (bounded by SO_RCVTIMEO=timeout_secs).
122// LATENCY = time to FIRST byte (ttfb) -- the real server-think time, IMMUNE to keep-alive/idle-close
123// artifacts (a daemon that sends fast then holds the socket open shows fast ttfb + slow total). Drains to
124// CLOSE (never closes early) so it can't SIGPIPE-kill a SIGPIPE-naive daemon.
125// out[0]=bytes, out[1]=verdict, out[2]=total ms (to close). Returns ttfb ms (or total if NO byte = hang).
126func ac_probe(port: i64, path: *u8, timeout_secs: i64, ip: i64, out: *i64) -> i64 {
127    out[0] = 0; out[1] = AC_V_OK; out[2] = 0
128    let t0: i64 = sys_now_ms()
129    let fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
130    if fd < 0 { out[1] = AC_V_SOCK; return 0 }
131    sys_set_socket_timeout(fd, timeout_secs)                 // backstop: a hung daemon can't wedge us forever
132    let sa: *u8 = sys_mmap(16); ac_addr(sa, (ip>>24)&0xff, (ip>>16)&0xff, (ip>>8)&0xff, ip&0xff, port)   // ip packed a.b.c.d (0x7f000001=loopback)
133    if nx_connect_bounded(fd, sa, 16, NX_CONN_DEFAULT_MS) < 0 { sys_close(fd); let td: i64 = sys_now_ms() - t0; out[1] = AC_V_DOWN; out[2] = td; return td }
134    let req: *u8 = sys_mmap(512); let rl: i64 = ac_build_get(req, path)
135    let wr: i64 = sys_write(fd, req, rl)
136    if wr != rl { sys_close(fd); let ts: i64 = sys_now_ms() - t0; out[1] = AC_V_SEND; out[2] = ts; return ts }
137    let buf: *u8 = sys_mmap(AC_MAGIC_65536)
138    var total: i64 = 0; var ttfb: i64 = 0 - 1; var keep: i64 = 1
139    while keep == 1 {
140        let r: i64 = sys_read(fd, buf, AC_MAGIC_65536)                // reuse buffer, drain to close (SIGPIPE-safe)
141        if r <= 0 { keep = 0 } else { if ttfb < 0 { ttfb = sys_now_ms() - t0 } total = total + r }
142    }
143    sys_close(fd)
144    let tend: i64 = sys_now_ms() - t0
145    out[0] = total; out[2] = tend
146    if ttfb < 0 { out[1] = AC_V_TIMEOUT; return tend }       // NO first byte within timeout = real hang
147    return ttfb                                              // latency = time to FIRST byte
148}
149
150// ===== one concurrency burst per target ==========================
151// fire `conc` simultaneous probes; aggregate on TTFB; print a row; append a time-series row.
152// VERDICT is ttfb-based (real server latency); `slowclose` (fast first byte but slow close = keep-alive/
153// idle artifact) is reported as INFO and NEVER causes AMBER/RED -- that is the artifact-immunity.
154// out_stats: [p50,p95,max,oks,stalls,timeouts,downs,min,maxtotal,slowclose]. Returns the AC_R_* code.
155// ---- SOVEREIGN seg-store (NO tsv): latest-per-label metric record + apm-ids index ----
156func ac_line_eq(buf: *u8, start: i64, end: i64, label: *u8) -> i64 {
157    var i: i64 = start; var j: i64 = 0
158    while i < end { if buf[i] != label[j] { return 0 } i = i + 1; j = j + 1 }
159    if label[j] != (0 as u8) { return 0 }
160    return 1
161}
162func ac_seg_index(label: *u8) -> i64 {
163    let buf: *u8 = sys_mmap(AC_MAGIC_65536); var n: i64 = 0
164    let fd: i64 = sys_openat_rd("knowledge/store/apm-ids" as *u8)
165    if fd >= 0 { n = sys_read(fd, buf, AC_MAGIC_65536); sys_close(fd) }
166    var present: i64 = 0; var ls: i64 = 0; var i: i64 = 0
167    while i <= n { var nl: i64=0; if i>=n {nl=1} else { if buf[i]==(10 as u8){nl=1} } if nl==1 { if i>ls { if ac_line_eq(buf, ls, i, label)==1 { present=1 } } ls=i+1 } i=i+1 }
168    if present == 0 { let afd: i64 = sys_openat_append("knowledge/store/apm-ids" as *u8, 0x1a4); if afd >= 0 { let lb: *u8=sys_mmap(128); var lo: i64=ac_s(lb,0,label); lb[lo]=10 as u8; lo=lo+1; sys_write(afd, lb, lo); sys_close(afd) } }
169    return 0
170}
171func ac_burst(port: i64, path: *u8, label: *u8, conc: i64, timeout_secs: i64, epoch: i64, ip: i64, out_stats: *i64) -> i64 {
172    // SHARED results (MAP_SHARED survives fork) -- [ttfb, bytes, verd, total] per probe. (sys_mmap_shared,
173    // NOT temp files: sys_read_file mmaps 4 GiB/call and ballooned parent VA into fork ENOMEM.)
174    let res: *i64 = sys_mmap_shared(8 * 4 * conc) as *i64
175    var z: i64 = 0; while z < 4 * conc { res[z] = 0 - 1; z = z + 1 }
176    let pids: *i64 = sys_mmap(8 * (conc + 1)) as *i64    // track OUR probe children only
177    var k: i64 = 0
178    while k < conc {
179        let pid: i64 = sys_fork()
180        if pid == 0 {
181            let st: *i64 = sys_mmap(32) as *i64
182            let tt: i64 = ac_probe(port, path, timeout_secs, ip, st)
183            res[k*4] = tt; res[k*4 + 1] = st[0]; res[k*4 + 2] = st[1]; res[k*4 + 3] = st[2]
184            sys_exit(0)
185        }
186        pids[k] = pid
187        k = k + 1
188    }
189    // Reap EXACTLY our probe pids (wait4(-1) would steal a sibling mock/daemon child in a multi-child caller).
190    let stp: *i64 = sys_mmap(16) as *i64; var w: i64 = 0
191    while w < conc { sys_wait4(pids[w], stp, 0); w = w + 1 }
192
193    let ms: *i64 = sys_mmap(8 * (conc + 2)) as *i64
194    var nres: i64 = 0; var oks: i64 = 0; var downs: i64 = 0; var timeouts: i64 = 0; var stalls: i64 = 0; var errs: i64 = 0; var slowclose: i64 = 0; var maxtot: i64 = 0
195    var kk: i64 = 0
196    while kk < conc {
197        let tt: i64 = res[kk*4]; let bytes: i64 = res[kk*4 + 1]; let verd: i64 = res[kk*4 + 2]; let tot: i64 = res[kk*4 + 3]
198        if tt >= 0 { ms[nres] = tt; nres = nres + 1; if tt >= AC_STALL_MS { stalls = stalls + 1 } }   // stall = slow FIRST byte
199        if tot > maxtot { maxtot = tot }
200        if verd == AC_V_OK { if bytes > 0 { oks = oks + 1; if tot >= AC_STALL_MS { if tt < AC_STALL_MS { slowclose = slowclose + 1 } } } else { errs = errs + 1 } }
201        else { if verd == AC_V_DOWN { downs = downs + 1 }
202               else { if verd == AC_V_TIMEOUT { timeouts = timeouts + 1 } else { if verd >= 0 { errs = errs + 1 } } } }
203        kk = kk + 1
204    }
205    ac_sort(ms, nres)
206    var mn: i64 = 0; var p50: i64 = 0; var p95: i64 = 0; var mx: i64 = 0
207    if nres > 0 { mn = ms[0]; mx = ms[nres-1]; p50 = ms[nres/2]; var pi: i64 = (nres*95)/100; if pi >= nres { pi = nres - 1 } p95 = ms[pi] }
208
209    // ttfb-based verdict (slowclose is INFO only -> artifact-immune)
210    var code: i64 = AC_R_GREEN
211    if stalls > 0 { code = AC_R_AMBER }
212    if timeouts > 0 { code = AC_R_RED }
213    if errs > 0 { code = AC_R_RED }
214    if downs > 0 { code = AC_R_RED }
215    if downs >= conc { code = AC_R_DOWN }
216
217    out_stats[0] = p50; out_stats[1] = p95; out_stats[2] = mx; out_stats[3] = oks
218    out_stats[4] = stalls; out_stats[5] = timeouts; out_stats[6] = downs; out_stats[7] = mn
219    out_stats[8] = maxtot; out_stats[9] = slowclose
220
221    var vstr: *u8 = "GREEN" as *u8
222    if code == AC_R_AMBER { vstr = "AMBER" as *u8 }
223    if code == AC_R_RED { vstr = "RED  " as *u8 }
224    if code == AC_R_DOWN { vstr = "DOWN " as *u8 }
225    ac_w("[" as *u8); ac_w(vstr); ac_w("] " as *u8)
226    let lw: i64 = ac_slen(label); sys_write(1, label, lw); var pad: i64 = 18 - lw; while pad > 0 { sys_write(1, " " as *u8, 1); pad = pad - 1 }
227    ac_w(" :" as *u8); ac_wn(port)
228    ac_w("  ok=" as *u8); ac_wn(oks); ac_w("/" as *u8); ac_wn(conc)
229    ac_w("  ttfb p50=" as *u8); ac_wpad(p50, 5); ac_w(" p95=" as *u8); ac_wpad(p95, 6); ac_w(" max=" as *u8); ac_wpad(mx, 6); ac_w("ms" as *u8)
230    ac_w("  stall=" as *u8); ac_wn(stalls); ac_w(" to=" as *u8); ac_wn(timeouts); ac_w(" down=" as *u8); ac_wn(downs); ac_w(" slowclose=" as *u8); ac_wn(slowclose)
231    ac_w("\n" as *u8)
232
233    // SOVEREIGN seg-store record (latest-per-target, key:value; NO tsv) -- per-key file + apm-ids index
234    sys_mkdir("knowledge/store" as *u8, 0x1ff)
235    let sp: *u8 = sys_mmap(512); var spo: i64 = ac_s(sp, 0, "knowledge/store/apm-" as *u8); spo = ac_s(sp, spo, label); sp[spo]=0 as u8
236    let rb: *u8 = sys_mmap(AC_MAGIC_1024); var ro: i64 = 0
237    ro=ac_s(rb,ro,"label=" as *u8); ro=ac_s(rb,ro,label); rb[ro]=10 as u8; ro=ro+1
238    ro=ac_s(rb,ro,"epoch=" as *u8); ro=ac_n(rb,ro,epoch); rb[ro]=10 as u8; ro=ro+1
239    ro=ac_s(rb,ro,"port=" as *u8); ro=ac_n(rb,ro,port); rb[ro]=10 as u8; ro=ro+1
240    ro=ac_s(rb,ro,"conc=" as *u8); ro=ac_n(rb,ro,conc); rb[ro]=10 as u8; ro=ro+1
241    ro=ac_s(rb,ro,"oks=" as *u8); ro=ac_n(rb,ro,oks); rb[ro]=10 as u8; ro=ro+1
242    ro=ac_s(rb,ro,"ttfb_p50=" as *u8); ro=ac_n(rb,ro,p50); rb[ro]=10 as u8; ro=ro+1
243    ro=ac_s(rb,ro,"ttfb_max=" as *u8); ro=ac_n(rb,ro,mx); rb[ro]=10 as u8; ro=ro+1
244    ro=ac_s(rb,ro,"stalls=" as *u8); ro=ac_n(rb,ro,stalls); rb[ro]=10 as u8; ro=ro+1
245    ro=ac_s(rb,ro,"timeouts=" as *u8); ro=ac_n(rb,ro,timeouts); rb[ro]=10 as u8; ro=ro+1
246    ro=ac_s(rb,ro,"downs=" as *u8); ro=ac_n(rb,ro,downs); rb[ro]=10 as u8; ro=ro+1
247    ro=ac_s(rb,ro,"slowclose=" as *u8); ro=ac_n(rb,ro,slowclose); rb[ro]=10 as u8; ro=ro+1
248    let mfd: i64 = sys_openat_wr(sp, 0x1a4); if mfd >= 0 { sys_write(mfd, rb, ro); sys_close(mfd) }
249    ac_seg_index(label)
250    return code
251}
252
253// parse dotted "a.b.c.d" -> packed i64 (a<<24|b<<16|c<<8|d)
254func ac_parse_ip(s: *u8) -> i64 {
255    let oct: *i64 = sys_mmap(40) as *i64
256    var idx: i64 = 0; var v: i64 = 0; var i: i64 = 0
257    while idx < 4 {
258        let c: i64 = s[i] as i64
259        if c == 46 { oct[idx] = v; idx = idx + 1; v = 0; i = i + 1 }
260        else { if c >= 48 { if c <= 57 { v = v*10 + (c-48); i = i + 1 } else { oct[idx] = v; idx = 4 } } else { oct[idx] = v; idx = 4 } }
261    }
262    oct[3] = v
263    return ((oct[0]&0xff)<<24) | ((oct[1]&0xff)<<16) | ((oct[2]&0xff)<<8) | (oct[3]&0xff)
264}
265
266func ac_starts2(buf: *u8, ls: i64, end: i64, key: *u8) -> i64 { var m: i64=0; while key[m]!=(0 as u8) { if ls+m>=end { return 0 } if buf[ls+m]!=key[m] { return 0 } m=m+1 } return m }
267func ac_kv_int(buf: *u8, n: i64, key: *u8) -> i64 {
268    var ls: i64=0; var i: i64=0
269    while i <= n { var nl: i64=0; if i>=n{nl=1} else { if buf[i]==(10 as u8){nl=1} }
270        if nl==1 { let kl: i64=ac_starts2(buf,ls,i,key); if kl>0 { var v: i64=0; var p: i64=ls+kl; while p<i { let c: i64=buf[p] as i64; if c>=48 { if c<=57 { v=v*10+(c-48) } } p=p+1 } return v } ls=i+1 }
271        i=i+1 }
272    return 0
273}
274func ac_kv_str(buf: *u8, n: i64, key: *u8, out: *u8, outcap: i64) -> i64 {
275    var ls: i64=0; var i: i64=0
276    while i <= n { var nl: i64=0; if i>=n{nl=1} else { if buf[i]==(10 as u8){nl=1} }
277        if nl==1 { let kl: i64=ac_starts2(buf,ls,i,key); if kl>0 { var w: i64=0; var p: i64=ls+kl; while p<i { if w<outcap-1 { out[w]=buf[p]; w=w+1 } p=p+1 } out[w]=0 as u8; return w } ls=i+1 }
278        i=i+1 }
279    out[0]=0 as u8; return 0
280}
281// one-shot migrate legacy apm_targets.tsv -> seg-store config (apmtarget-<label> key:value + apmtarget-ids). idempotent.
282func ac_migrate_targets() -> i64 {
283    let chk: i64 = sys_openat_rd("knowledge/store/apmtarget-ids" as *u8); if chk >= 0 { sys_close(chk); return 0 }
284    let cfg: *u8 = sys_mmap(AC_MAGIC_65536); let n: i64 = ac_read_small(AC_TARGETS, cfg, AC_MAGIC_65536); if n <= 0 { return 0 }
285    sys_mkdir("knowledge/store" as *u8, 0x1ff)
286    let f: *u8 = sys_mmap(256); let path: *u8 = sys_mmap(256); let label: *u8 = sys_mmap(64); let stp: *i64 = sys_mmap(8) as *i64
287    var i: i64 = 0
288    while i < n {
289        if cfg[i] == (35 as u8) { var g: i64 = 1; while g == 1 { if i >= n { g = 0 } else { if cfg[i] == (10 as u8) { i = i + 1; g = 0 } else { i = i + 1 } } } }
290        else { if cfg[i] == (10 as u8) { i = i + 1 }
291        else {
292            i = ac_field(cfg, n, i, f, 256, stp); let port: i64 = ac_atoi(f)
293            if stp[0] == 0 {
294                i = ac_field(cfg, n, i, path, 256, stp)
295                if stp[0] == 0 {
296                    i = ac_field(cfg, n, i, label, 64, stp)
297                    var conc: i64 = 8
298                    if stp[0] == 0 { i = ac_field(cfg, n, i, f, 256, stp); conc = ac_atoi(f) }
299                    if conc <= 0 { conc = 8 }
300                    if port > 0 {
301                        let sp: *u8 = sys_mmap(512); var so: i64 = ac_s(sp, 0, "knowledge/store/apmtarget-" as *u8); so = ac_s(sp, so, label); sp[so] = 0 as u8
302                        let rb: *u8 = sys_mmap(512); var ro: i64 = 0
303                        ro = ac_s(rb, ro, "port=" as *u8); ro = ac_n(rb, ro, port); rb[ro] = 10 as u8; ro = ro + 1
304                        ro = ac_s(rb, ro, "path=" as *u8); ro = ac_s(rb, ro, path); rb[ro] = 10 as u8; ro = ro + 1
305                        ro = ac_s(rb, ro, "conc=" as *u8); ro = ac_n(rb, ro, conc); rb[ro] = 10 as u8; ro = ro + 1
306                        let fd: i64 = sys_openat_wr(sp, 0x1a4); if fd >= 0 { sys_write(fd, rb, ro); sys_close(fd) }
307                        let il: *u8 = sys_mmap(128); var io: i64 = ac_s(il, 0, label); il[io] = 10 as u8; io = io + 1
308                        let ifd: i64 = sys_openat_append("knowledge/store/apmtarget-ids" as *u8, 0x1a4); if ifd >= 0 { sys_write(ifd, il, io); sys_close(ifd) }
309                    }
310                } else { if stp[0] == 1 { i = i } }
311            }
312        } }
313    }
314    return 0
315}
316
317func main(argc: i64, argv: *i64) -> i64 {
318    let epoch: i64 = sys_now_realtime_sec()
319    var target_ip: i64 = 0x7f000001                          // default loopback (on-NAS); argv[1]=a.b.c.d to probe over LAN
320    if argc >= 2 { target_ip = ac_parse_ip(argv[1] as *u8) }
321    ac_w("=== NISHI APM COLLECT (HTTP, Connection: close, no TLS; latency=TTFB) ip=" as *u8); ac_wn((target_ip>>24)&0xff); ac_w("." as *u8); ac_wn((target_ip>>16)&0xff); ac_w("." as *u8); ac_wn((target_ip>>8)&0xff); ac_w("." as *u8); ac_wn(target_ip&0xff); ac_w(" ===\n" as *u8)
322    ac_w("    stall=ttfb>2000ms  timeout=" as *u8); ac_wn(AC_TIMEOUT_S); ac_w("s  GREEN/AMBER(slow first byte)/RED(no response|err)/DOWN(refused); slowclose=keep-alive artifact (info)\n\n" as *u8)
323    ac_migrate_targets()
324    let idxb: *u8 = sys_mmap(AC_MAGIC_65536); let idn: i64 = ac_read_small("knowledge/store/apmtarget-ids" as *u8, idxb, AC_MAGIC_65536)
325    if idn <= 0 { ac_w("no targets: seg-store knowledge/store/apmtarget-ids empty and no legacy apm_targets.tsv to migrate\n" as *u8); return 0 }
326    let st: *i64 = sys_mmap(8 * 12) as *i64
327    let label: *u8 = sys_mmap(64); let path: *u8 = sys_mmap(256); let recp: *u8 = sys_mmap(512); let rec: *u8 = sys_mmap(AC_MAGIC_1024)
328    var ntargets: i64 = 0; var ls: i64 = 0; var i: i64 = 0
329    while i <= idn {
330        var nl: i64 = 0; if i >= idn { nl = 1 } else { if idxb[i] == (10 as u8) { nl = 1 } }
331        if nl == 1 {
332            if i > ls {
333                var w: i64 = 0; var k: i64 = ls; while k < i { if w < 63 { label[w] = idxb[k]; w = w + 1 } k = k + 1 } label[w] = 0 as u8
334                var po: i64 = ac_s(recp, 0, "knowledge/store/apmtarget-" as *u8); po = ac_s(recp, po, label); recp[po] = 0 as u8
335                let rn: i64 = ac_read_small(recp, rec, AC_MAGIC_1024)
336                if rn > 0 {
337                    let port: i64 = ac_kv_int(rec, rn, "port=" as *u8)
338                    ac_kv_str(rec, rn, "path=" as *u8, path, 256)
339                    var conc: i64 = ac_kv_int(rec, rn, "conc=" as *u8); if conc <= 0 { conc = 8 }
340                    if port > 0 { ac_burst(port, path, label, conc, AC_TIMEOUT_S, epoch, target_ip, st); ntargets = ntargets + 1 }
341                }
342            }
343            ls = i + 1
344        }
345        i = i + 1
346    }
347    ac_w("\ncollected " as *u8); ac_wn(ntargets); ac_w(" targets -> seg-store knowledge/store/apm-* (epoch=" as *u8); ac_wn(epoch); ac_w(")\n" as *u8)
348    return 0
349}
nx_apm_collect.nx source

`nx_apm_collect.nx` source
