diff --git a/di/clienttracking/VERSION b/di/clienttracking/VERSION new file mode 100644 index 00000000..6e8bf73a --- /dev/null +++ b/di/clienttracking/VERSION @@ -0,0 +1 @@ +0.1.0 diff --git a/di/clienttracking/clienttracking.md b/di/clienttracking/clienttracking.md new file mode 100644 index 00000000..6cd4847c --- /dev/null +++ b/di/clienttracking/clienttracking.md @@ -0,0 +1,235 @@ +# di.clienttracking + +Tracks the client sessions connected to a KDB-X process in an in-memory session table — who +connected, when they opened and closed, and (once a query owner exists) how many requests and how +many result-bytes each client has been served. It is the modular replacement for TorQ's +`code/handlers/trackclients.q` (`.clients` namespace). + +It never assigns `.z.*` directly: all connection-lifecycle and query hooks are registered through an +injected **di.handlers** instance, so it coexists with any other component hooking the same events. + +--- + +## Dependencies + +| Dependency | Key | Required | Description | +|---|---|---|---| +| logger | `` `log `` | yes | dict with `info`, `warn`, `error`, each binary `{[c;m]}` (context symbol, message string) | +| handlers | `` `handlers `` | yes | dict with `register`, `remove`, `list` — a di.handlers instance's exported functions | + +Both are **injected via `init`**. `di.clienttracking` has **no hard `use` dependency** on another +`di.*` module. + +> **Note on the tier classification.** The modularisation plan's dependency tree lists +> `di.clienttracking → di.handlers`. That is a *role/coupling* grouping (client-tracking is +> meaningless without a handler registry), **not** a hard `use` import. Principle 4 and the injectable +> contract table both class handler management as an *injected* dependency, so — consistent with the +> rest of the framework — di.handlers is passed in as `deps[`handlers]`, and this module loads and +> tests standalone. `di.torq` wires the shared, already-`init`-ed di.handlers instance into this +> module's `init` at startup. + +`init` throws immediately if either dependency is missing, is not a dict, or is missing a required +key. No adaptation is performed — pass dicts that already conform. + +--- + +## Initialisation + +```q +handlers:use`di.handlers +ct:use`di.clienttracking +logger:use`di.log + +/ di.log supplies the log dependency - `logdict`log is an info..fatal level dict +/ (a superset of the required info/warn/error); it is passed straight through, no adaptation +logdep:logger.logdict`log + +/ di.handlers must be initialised before it is handed on +handlers.init[enlist[`log]!enlist logdep] +hdep:`register`remove`list!(handlers.register;handlers.remove;handlers.list) + +ct.init[`log`handlers!(logdep;hdep)] +``` + +Any dict with binary `` `info`warn`error `` `{[c;m]}` functions works — a hand-rolled one for a quick +test, or `di.log`'s `logdict`log` in a real process. + +`init` must be called before any other function (there is no default logger). It is idempotent — +re-calling re-wires the dependencies, re-registers the lifecycle handlers in place, and preserves the +existing session table. Re-calling with `trackusage:0b` also *removes* any usage handlers a previous +`init`/`enableusage` had wired, so the flag is authoritative on every call. + +### Config keys (optional) + +Each is type-checked by `init` (a wrong type is rejected up front, not deferred to a runtime failure): + +| Key | Type | Default | Meaning | +|---|---|---|---| +| `maxidle` | timespan | `0D00:15:00` | force-close a live handle idle longer than this; `0D` disables idle reaping | +| `retain` | timespan | `0D00:05:00` | purge a closed session this long after it ended | +| `trackusage` | boolean | `1b` | wire usage counting during `init` (`0b` also tears down any already-wired usage handlers) | + +--- + +## The session table + +`getclients[]` returns the table; one row per session, an open session has a null `endp`: + +| Column | Type | Meaning | +|---|---|---| +| `w` | `` `g#int `` | connection handle (`.z.w` at open) | +| `ipa` | symbol | client ip address, dotted-decimal | +| `u` | symbol | client user (`.z.u` at open) | +| `a` | int | client ip address, raw int (`.z.a` at open) | +| `startp` | timestamp | session start — connection open time | +| `endp` | timestamp | session end — connection close time; null while open | +| `lastp` | timestamp | time of the last request seen from this client | +| `hits` | long | number of requests served for this client | +| `sz` | long | total (approximate, via `-22!`) bytes of results returned | + +Current connections are `select from ct.getclients[] where null endp`. + +--- + +## Exported functions + +### `init[deps]` +Wire dependencies + config, create the session table, register the lifecycle handlers, and (if +`trackusage`) attempt to wire usage counting. Idempotent. + +### `getclients[]` +Return the session table. + +### `addclient[handle]` +Manually record a client `handle` (an int) as an open session, using the current `.z` context — the +equivalent of TorQ's `addw`. Signals if `handle` is not an int. + +### `cleanup[]` +Run a cleanup sweep now: stamp `endp` on open rows whose handle is no longer live, force-close live +handles idle past `maxidle`, and delete closed rows older than `retain`. Cleanup also runs +automatically on every connection open and close; export it so a host can also drive it from a +`di.timer` job. + +### `enableusage[]` +(Re)wire usage counting onto each of `.z.pg` / `.z.ps` / `.z.ws` that currently has an `exec` owner. +Idempotent. Call it after the process's query owner (e.g. a gateway or `di.permissions`) has been +registered — see the design note below on why usage counting is deferred. + +### `version` +The module version string (`"0.1.0"`). + +--- + +## Events managed + +| Event | di.handlers model | Role here | +|---|---|---| +| `.z.po` | simple (observer) | open a session row | +| `.z.pc` | simple (observer) | stamp `endp` on the session | +| `.z.wo` | simple (observer) | open a session row (websocket) | +| `.z.wc` | simple (observer) | stamp `endp` on the session (websocket) | +| `.z.pg` | phased — `post` | count a served request (hits, bytes, lastp) | +| `.z.ps` | phased — `post` | count a served request | +| `.z.ws` | phased — `post` | count a served request | + +All registrations use the name `` `clienttracking `` at priority `0`. + +--- + +## Design decisions & rationale + +The extraction turned on how di.handlers classifies events, and several deliberate departures from +TorQ's `trackclients.q` follow from it. + +- **Lifecycle vs usage are two different di.handlers models.** Connection open/close (`.z.po`/`.z.pc` + /`.z.wo`/`.z.wc`) are *simple* events: di.handlers fans them out to every registrant and discards + the return, so client-tracking is just one more observer. Per-request counting hangs off + `.z.pg`/`.z.ps`/`.z.ws`, which are *phased* — their return value is the query answer. Counting is a + side-effect watcher of a result, i.e. a **`post`** handler, never an owner or a `pre`. + +- **Usage counting is deferred, by necessity.** di.handlers refuses a `post` registration until an + `exec` owner exists on that event (the dispatcher isn't installed until then). So `init` wires the + four lifecycle observers unconditionally, then *attempts* usage counting: for each query event that + already has an owner it registers the `post` handler; for the rest it logs a `warn` and skips. + `enableusage[]` re-attempts and is meant to be called once the query owner is up. This ordering + dependency is inherent to the observer/decider split — it is surfaced, not hidden. + +- **`errs` is not tracked (deliberate omission).** TorQ counted per-client errors via a wrapper that + saw the failure. In di.handlers a `post` handler runs **only after a successful `exec`** — a throwing + query propagates before any `post` fires — so a `post` watcher structurally cannot observe errors. + Rather than carry a column that would always be zero, the `errs` column is dropped. Error attribution + would need a different mechanism (e.g. an owner/`pre` that traps) and is out of scope here. + +- **INTRUSIVE mode dropped.** TorQ optionally sent an async `eval` back to each connecting client to + self-report its `.z.k`/`.z.c`/os/pid/port. It only works for a cooperating q client and is + security-questionable; it is removed, and with it the `k`/`K`/`c`/`s`/`o`/`f`/`pid`/`port` columns + it populated. The retained schema is what this module can populate truthfully from `.z.*` at + connect time. + +- **Unkeyed session table.** TorQ keyed `CLIENTS` on the handle and nulled the key on close, which + collides when handles are reused or several closed rows coexist. This module keeps an unkeyed table + with `` `g# `` on `w`; the current session for a handle is the row with that `w` and a null `endp`. + Handle reuse simply produces a new row. + +- **Cleanup is inline, no timer injected.** Cleanup runs on every open/close (as in TorQ) and is also + exported for a host to schedule. That keeps the injectable surface to `log` + `handlers` only; a + host wanting periodic sweeps wires `cleanup[]` to `di.timer` rather than this module pulling timer + in. + +## Known limitations + +- **Usage counting needs a single-threaded query port.** A `post` handler on `.z.pg`/`.z.ps`/`.z.ws` + runs in the query's own execution context. On a multithreaded process (a negative `\p` port) that is + not the main thread, so the global write inside the counter hits kdb's `'noupdate` restriction — the + same constraint any `.z.pg` code faces. di.handlers isolates the `post`, so the query still succeeds; + the count is simply skipped and a `warn` is logged. Lifecycle tracking (open/close) is unaffected — + those run on the main thread. +- **Usage counting is order-dependent.** It only attaches to query events that already have an `exec` + owner. If a query owner is registered *after* `di.clienttracking`, call `enableusage[]` again. If the + owner is later removed, di.handlers clears the event's `post` state with it — re-run `enableusage[]` + after re-registering an owner. +- **`ipa` is a plain dotted-decimal format** of `.z.a`; unlike TorQ it does not do a reverse-hostname + lookup or cache. + +--- + +## Running tests + +Needs KDB-X (the `use` module system + k4unit). + +**Unit suite** (`test.csv`, 44 checks) — hermetic, no sockets. Runs against the **real, merged +di.handlers and di.log** (nothing is mocked); drives di.handlers' actual dispatcher by invoking the +function bound to each `.z.*` event with synthetic handles. `moduletest` loads and runs it: + +```q +k4unit:use`di.k4unit +k4unit.moduletest`di.clienttracking / "All tests passed" +``` + +It covers dependency + config-type validation, the four lifecycle registrations, open/close through +real di.handlers dispatch (which also proves the registered callbacks resolve this module's own `.z.m` +state), `addclient`, cleanup of dead handles, usage-counting deferral without an owner and activation +with one, teardown on `trackusage:0b` re-init, and the api-metadata/version contract. + +**Integration suite** (`test_integration.csv`, 6 checks) — stands up a real child q process, tracks the +outgoing handle to it, then reaps it as idle. This is the one path the unit suite cannot reach: the +idle-reap branch force-closes a handle that must genuinely be in `.z.W`. It needs a q/kdb-x binary via +`QHOME`, needs no other configuration, and skips cleanly if none is available. `moduletest` only loads +`test.csv`, so run this suite directly, in a fresh session: + +```q +k4unit:use`di.k4unit +.m.di.0k4unit.KUltf .Q.dd[hsym`$.Q.m.mp`di.clienttracking;`test_integration.csv] +.m.di.0k4unit.KUrt[] +k4unit.getresults[] / one row per assertion; ok=1 is a pass +``` + +The *incoming*-connection (`.z.po`) direction is deliberately not integration-tested: a process only +accepts an inbound connection at its top-level event loop, which a k4unit script never reaches. That +di.handlers binds `.z.*` to the socket layer at all is covered by di.handlers' own integration suite; +here the `.z.po`/`.z.pc` dispatch is exercised via the unit suite's synthetic-handle drives. + +```q +k4unit:use`di.k4unit +k4unit.moduletest`di.clienttracking / prints the results; "All tests passed" on success +``` diff --git a/di/clienttracking/clienttracking.q b/di/clienttracking/clienttracking.q new file mode 100644 index 00000000..78cc6d0e --- /dev/null +++ b/di/clienttracking/clienttracking.q @@ -0,0 +1,175 @@ +/ track the client sessions connected to a KDB-X process in a session table +/ consumes di.handlers (injected) to observe connection open/close and, once a query owner +/ exists, to count per-request usage - it never assigns .z.* directly +/ NB the exported `version` is defined in init.q (read from the VERSION file), not here + +/ ============================================================ +/ constants (load-time) +/ ============================================================ + +/ session table schema - one row per client session; an open session has a null endp +/ w is grouped for fast per-handle lookup; hits/sz are longs +clientschema:([] + w:`g#`int$(); / connection handle (.z.w at open) + ipa:`symbol$(); / client ip address, dotted-decimal + u:`symbol$(); / client user (.z.u at open) + a:`int$(); / client ip address, raw int (.z.a at open) + startp:`timestamp$(); / session start - connection open time + endp:`timestamp$(); / session end - connection close time; null while open + lastp:`timestamp$(); / time of the last request seen from this client + hits:`long$(); / number of requests served for this client + sz:`long$()); / total (approximate) bytes of results returned to this client + +/ priority used when registering with di.handlers - lower runs first; 0 is a neutral default +handlerpriority:0; + +/ idle/retention defaults, overridable via init config +defaultmaxidle:0D00:15:00; / force-close a live handle idle for longer than this (0D disables) +defaultretain:0D00:05:00; / purge a closed session this long after it ended + +/ the phased query events usage counting attaches to (as post-phase watchers) +usageevents:`.z.pg`.z.ps`.z.ws; + +/ ============================================================ +/ internal helpers +/ ============================================================ + +ipa:{[a] + / format a raw .z.a int ip address as a dotted-decimal symbol + / .z.a is a signed int32 (high ips are negative); 0x0 vs takes its two's-complement bytes and + / "i"$ casts each byte UNSIGNED (0-255), so octets >= 128 render correctly (e.g. 192.168.0.1) + :`$"." sv string "i"$0x0 vs a; + }; + +raiseerror:{[ctx;msg] + / internal - log an error under ctx then signal it, so failures are observable as well as thrown + .z.m.logerr[ctx;msg]; + '"di.clienttracking: ",string[ctx],": ",msg; + }; + +track:{[h] + / record a newly-seen client handle - sweep first, then append an open session row for h + cleanup[]; + .z.m.clients:.z.m.clients upsert (h;ipa .z.a;.z.u;.z.a;.z.p;0Np;.z.p;0j;0j); + }; + +closeclient:{[h] + / mark the open session for handle h as closed, then sweep + .z.m.clients:update endp:.z.p from .z.m.clients where w=h,null endp; + cleanup[]; + }; + +hitpost:{[result;args] + / usage post-handler - di.handlers calls post[result;args], so this must be binary; args is unused + / bumps the request count and result-byte total for the calling client (.z.w) + / runs only on a successful exec (di.handlers post fires after the owner returns), so it cannot + / see errors; on a multithreaded (negative \p) process a global write here hits 'noupdate and is + / isolated/logged by di.handlers rather than counting + .z.m.clients:update lastp:.z.p,hits:hits+1,sz:sz+-22!result from .z.m.clients where w=.z.w,null endp; + }; + +disableusage:{[] + / remove any usage post-handlers this module registered (idempotent - di.handlers no-ops an unknown name) + .z.m.handlers[`remove][;`post;`clienttracking] each usageevents; + }; + +registerlifecycle:{[] + / register the connection and websocket open/close observers (idempotent - di.handlers replaces in place) + .z.m.handlers[`register][`.z.po;`;`clienttracking;handlerpriority;track]; + .z.m.handlers[`register][`.z.pc;`;`clienttracking;handlerpriority;closeclient]; + .z.m.handlers[`register][`.z.wo;`;`clienttracking;handlerpriority;track]; + .z.m.handlers[`register][`.z.wc;`;`clienttracking;handlerpriority;closeclient]; + }; + +/ ============================================================ +/ public api +/ ============================================================ + +init:{[deps] + / wire the required dependencies (log, handlers) and optional config, then register the lifecycle handlers + / deps: a dict with `log (info/warn/error binary {[c;m]} funcs) and `handlers (di.handlers register/remove/list) + / optional: `maxidle (timespan), `retain (timespan), `trackusage (boolean, default 1b) + / example: ct.init[`log`handlers!(logdep;hdep)] + if[99h<>type deps; + '"di.clienttracking: deps must be a dict with `log and `handlers keys"]; + if[not `log in key deps; + '"di.clienttracking: log dependency is required; pass `info`warn`error functions keyed on `log"]; + if[99h<>type deps`log; + '"di.clienttracking: log value must be a dict; pass `info`warn`error functions"]; + if[not all `info`warn`error in key deps`log; + '"di.clienttracking: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)]; + if[not `handlers in key deps; + '"di.clienttracking: handlers dependency is required; pass di.handlers register/remove/list keyed on `handlers"]; + if[99h<>type deps`handlers; + '"di.clienttracking: handlers value must be a dict; pass register/remove/list functions"]; + if[not all `register`remove`list in key deps`handlers; + '"di.clienttracking: handlers dict must have `register`remove`list keys; got: ",(", " sv string key deps`handlers)]; + / optional config - validate types up front (nested if, not `and`, to avoid eager-eval on an absent key) + if[`maxidle in key deps; + if[not -16h=type deps`maxidle;'"di.clienttracking: maxidle must be a timespan"]]; + if[`retain in key deps; + if[not -16h=type deps`retain;'"di.clienttracking: retain must be a timespan"]]; + if[`trackusage in key deps; + if[not -1h=type deps`trackusage;'"di.clienttracking: trackusage must be a boolean"]]; + .z.m.loginfo:(deps`log)`info; + .z.m.logwarn:(deps`log)`warn; + .z.m.logerr:(deps`log)`error; + .z.m.handlers:deps`handlers; + .z.m.maxidle:$[`maxidle in key deps;deps`maxidle;defaultmaxidle]; + .z.m.retain:$[`retain in key deps;deps`retain;defaultretain]; + .z.m.trackusage:$[`trackusage in key deps;deps`trackusage;1b]; + / create the session table on first init only; a re-init must leave the existing table intact + if[not `clients in key .z.m;.z.m.clients:clientschema]; + registerlifecycle[]; + $[.z.m.trackusage;enableusage[];disableusage[]]; + .z.m.loginfo[`init;"di.clienttracking initialised"]; + }; + +getclients:{[] + / the current client-session tracking table - one row per session (open rows have a null endp) + :.z.m.clients; + }; + +addclient:{[h] + / manually record a client handle in the tracking table using the current .z context (like TorQ's addw) + if[not -6h=type h;raiseerror[`addclient;"handle must be an int"]]; + track h; + }; + +cleanup:{[] + / reap sessions whose handle has gone, force-close idle live handles, purge expired closed rows + / runs automatically on every open/close; also exported so a host can drive it periodically (di.timer) + now:.z.p; + .z.m.clients:update endp:now from .z.m.clients where null endp,not w in key .z.W; + if[0D<.z.m.maxidle; + idle:exec w from .z.m.clients where null endp,w in key .z.W,lastp minimum version string. +/ di.clienttracking has NO hard `use` dependencies. di.handlers is a process-wide singleton (it owns +/ the one root .z.* per process) so it cannot be `use`d per-consumer - di.torq loads it once and +/ INJECTS it via init, along with di.log. both injected deps are audited by di.depcheck's core- +/ dependency-contract check, not declared here (mirroring di.depcheck's own empty deps.q). +deps:(`$())!(); diff --git a/di/clienttracking/init.q b/di/clienttracking/init.q new file mode 100644 index 00000000..7962173f --- /dev/null +++ b/di/clienttracking/init.q @@ -0,0 +1,7 @@ +/ di.clienttracking - track connected client sessions via di.handlers connection-lifecycle events +\l ::clienttracking.q +/ module version, read from the VERSION file (one plain-text file to bump per release). read +/ module-relative at load (`:::` resolves to di/clienttracking) and BEFORE export, since export:([...]) +/ evaluates each name; version stays in the export so di.depcheck reads it from the export dict +version:trim first read0`:::VERSION +export:([init;getclients;addclient;cleanup;enableusage;getapimeta;version]) diff --git a/di/clienttracking/test.csv b/di/clienttracking/test.csv new file mode 100644 index 00000000..b92e67b3 --- /dev/null +++ b/di/clienttracking/test.csv @@ -0,0 +1,77 @@ +action,ms,bytes,lang,code,repeat,minver,comment +comment,,,,,,,setup - load di.handlers + di.clienttracking and init against the real di.log + di.handlers +before,0,0,q,handlers:use`di.handlers,1,1,load the real di.handlers module +before,0,0,q,ct:use`di.clienttracking,1,1,load di.clienttracking module +before,0,0,q,logger:use`di.log,1,1,load the real di.log module +before,0,0,q,logdep:logger.logdict`log,1,1,di.log's info..fatal level dict (superset of info/warn/error) +before,0,0,q,handlers.init[enlist[`log]!enlist logdep],1,1,init di.handlers with the real di.log +before,0,0,q,"hdep:`register`remove`list!(handlers.register;handlers.remove;handlers.list)",1,1,build the handlers injectable dict +before,0,0,q,"ct.init[`log`handlers!(logdep;hdep)]",1,1,init di.clienttracking with the real log + handlers deps + +comment,,,,,,,init - dependency validation +fail,0,0,q,ct.init[(::)],1,1,init rejects a non-dict deps +fail,0,0,q,ct.init[()!()],1,1,init rejects missing log key +fail,0,0,q,ct.init[enlist[`log]!enlist 42],1,1,init rejects a non-dict log value +fail,0,0,q,"ct.init[enlist[`log]!enlist `info`warn!(logdep`info;logdep`warn)]",1,1,init rejects a log dict missing the error key +fail,0,0,q,ct.init[enlist[`log]!enlist logdep],1,1,init rejects missing handlers key +fail,0,0,q,"ct.init[`log`handlers!(logdep;42)]",1,1,init rejects a non-dict handlers value +fail,0,0,q,"ct.init[`log`handlers!(logdep;enlist[`register]!enlist{})]",1,1,init rejects a handlers dict missing remove/list keys +fail,0,0,q,"ct.init[`log`handlers`maxidle!(logdep;hdep;5)]",1,1,init rejects a non-timespan maxidle +fail,0,0,q,"ct.init[`log`handlers`trackusage!(logdep;hdep;1)]",1,1,init rejects a non-boolean trackusage +run,0,0,q,".ct.errstr:@[{ct.init[(::)]};(::);{x}]",1,1,capture the error string from a bad init +true,0,0,q,.ct.errstr like "di.clienttracking:*",1,1,init error is prefixed di.clienttracking: +run,0,0,q,"ct.init[`log`handlers!(logdep;hdep)]",1,1,re-init cleanly for the remaining tests + +comment,,,,,,,module metadata - exported version +true,0,0,q,10h=type ct.version,1,1,version is a string +true,0,0,q,0/dev/null 2>&1 & echo $!");h:.it.wait[p;20];ok:$[null h;0b;pid=@[h;".z.i";{0N}]];$[ok;(h;pid);[if[not null h;@[hclose;h;::]];@[{system "" sv ("kill -9 ";string pid)};::;{}];.it.spawn[qbin;n-1]]]},1,1,bind a free port and confirm via PID we reached our own child (retries past a stolen port) +before,0,0,q,sp:.it.spawn[qbin;5],1,1,spawn the child with up to five attempts +before,0,0,q,h:sp 0,1,1,the confirmed outgoing handle to the child +before,0,0,q,cpid:sp 1,1,1,the confirmed child PID +before,0,0,q,if[null h;exit 0],1,1,skip cleanly if no confirmed child came up +before,0,0,q,@[{h ".z.pc:{exit 0}"};::;{x}],1,1,backstop - child self-exits when its connection drops +before,0,0,q,roundtrip:@[{4=h"2+2"};::;{0b}],1,1,a real synchronous round-trip works +before,0,0,q,ct.addclient[h],1,1,track the real live handle +before,0,0,q,"tracked:0