accesslog-shipper: emit __shipstat__ sent-line checkpoints for splits reconciliation
The splits completeness floor is TIME-based (trust hours complete since shipper
connect) and blind to event-loss-by-rotation (2026-07-10: raw_events rotated
every 1-2min, cruncher lost ~85% of us-32 events, floor still called the hours
complete). Shipper now injects {"__shipstat__":1,"sess":..,"sent":N,"ts":..}
every 1000 lines or 30s through the SAME tcp pipe. Cruncher diffs shipper-sent
vs DB-ingested per gateway/hour -> a reconciliation-based floor that catches the
loss a clock cannot. sess=connection-start epoch so resets on reconnect are
detectable. Logic in a mounted .awk file (not inline) to dodge sh/awk quoting.
Verified on busybox:1.36 (systime/fflush OK). NOT deployed until claude@'s
cruncher filters __shipstat__ from event parsing - coordinating via SIP.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
18
accesslog-shipper.awk
Normal file
18
accesslog-shipper.awk
Normal file
@@ -0,0 +1,18 @@
|
|||||||
|
# Streams dshackle access-log lines unchanged, and every 1000 lines OR 30s
|
||||||
|
# injects a __shipstat__ checkpoint carrying the cumulative sent-count for this
|
||||||
|
# connection. The splits cruncher (claude host) filters these OUT of event
|
||||||
|
# processing and uses them to reconcile shipped-vs-ingested per gateway per hour
|
||||||
|
# — the completeness signal the time-based floor cannot see (event-loss-by-
|
||||||
|
# rotation, 2026-07-10). `sess` = connection-start epoch so the cruncher detects
|
||||||
|
# counter resets on shipper reconnect (retry loop resumes at file END).
|
||||||
|
BEGIN { sess = systime(); last = sess; n = 0 }
|
||||||
|
{
|
||||||
|
n++
|
||||||
|
print
|
||||||
|
now = systime()
|
||||||
|
if (n % 1000 == 0 || now - last >= 30) {
|
||||||
|
printf "{\"__shipstat__\":1,\"sess\":%d,\"sent\":%d,\"ts\":%d}\n", sess, n, now
|
||||||
|
fflush()
|
||||||
|
last = now
|
||||||
|
}
|
||||||
|
}
|
||||||
7
drpc.yml
7
drpc.yml
@@ -11,9 +11,14 @@ services:
|
|||||||
restart: unless-stopped
|
restart: unless-stopped
|
||||||
volumes:
|
volumes:
|
||||||
- ./dshackle_logs:/logs:ro
|
- ./dshackle_logs:/logs:ro
|
||||||
|
# sent-line counter: injects __shipstat__ checkpoints into the stream so
|
||||||
|
# the listener cruncher can reconcile shipped-vs-ingested per gateway/hour
|
||||||
|
# (fixes the time-based completeness floor being blind to event-loss-by-
|
||||||
|
# rotation, 2026-07-10). Cruncher MUST filter __shipstat__ from events.
|
||||||
|
- ./accesslog-shipper.awk:/shipper.awk:ro
|
||||||
command: >-
|
command: >-
|
||||||
sh -c 'while true; do
|
sh -c 'while true; do
|
||||||
tail -F /logs/access_log.jsonl 2>/dev/null | nc ${SPLITS_LISTENER_HOST:-78.47.193.225} ${SPLITS_LISTENER_PORT:-9102};
|
tail -F /logs/access_log.jsonl 2>/dev/null | awk -f /shipper.awk | nc ${SPLITS_LISTENER_HOST:-78.47.193.225} ${SPLITS_LISTENER_PORT:-9102};
|
||||||
echo shipper lost listener connection, retrying in 10s;
|
echo shipper lost listener connection, retrying in 10s;
|
||||||
sleep 10; done'
|
sleep 10; done'
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user