From 95e7317b812884aa7916f1dce7047a089204fdca Mon Sep 17 00:00:00 2001 From: Matt Kulka Date: Thu, 22 Oct 2026 18:48:25 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20log=20aggregation=20=E2=80=94=20the=20m?= =?UTF-8?q?achine=20ships=20its=20logs=20to=20Loki?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gameplay logs sat on /dump on the machine, readable only over SSH. Now they land in Loki for query and tail: - k8s/loki/loki.yaml: Loki 3.4.4 monolithic on the home cluster's monitoring namespace — filesystem storage on CephFS, 30-day retention, traefik ingress at loki.k8s-home.lqx.net with a cert-manager cert, Grafana datasource auto-registered via the sidecar. reject_old_samples_max_age is a year: the machines backfill their history on first ship. - shipper/: logshipper, a ~300-line static Go service for the machine. Tails /dump/mode.log and /dump/switch.log, batches, pushes to Loki's push API. Per-file offsets are saved only after a successful push (down Loki = retry-then-catch-up, never loss); first run starts at each file's end, SHIP_FROM_START=1 ships history; truncation is noticed. Lines keep their own RFC 3339 stamp as the Loki timestamp; unstamped lines (the boot banner) inherit the next stamped line's, so a stream never starts at "now" and jumps back. After a backfill it calls /flush — old entries are invisible in the ingester's head until they reach the store (found the hard way: Loki answers 204 to an old entry on a new stream and then never serves it). Labels: job=pinball, machine, source, game (from the boot line). /etc/init.d/log_shipper runs it at boot. Deployed: Loki is live and verified (query through the ingress). The shipper is verified end-to-end against the cluster from this Mac; the machine was unreachable at deploy time — the README has the install commands. Co-Authored-By: Claude Code --- .gitignore | 1 + README.md | 54 ++++++++- k8s/loki/loki.yaml | 163 ++++++++++++++++++++++++++ shipper/build.sh | 7 ++ shipper/go.mod | 3 + shipper/init.d/log_shipper | 21 ++++ shipper/main.go | 221 +++++++++++++++++++++++++++++++++++ shipper/shipper.go | 230 +++++++++++++++++++++++++++++++++++++ shipper/shipper_test.go | 191 ++++++++++++++++++++++++++++++ 9 files changed, 885 insertions(+), 6 deletions(-) create mode 100644 k8s/loki/loki.yaml create mode 100755 shipper/build.sh create mode 100644 shipper/go.mod create mode 100755 shipper/init.d/log_shipper create mode 100644 shipper/main.go create mode 100644 shipper/shipper.go create mode 100644 shipper/shipper_test.go diff --git a/.gitignore b/.gitignore index 68cb0be..af2b3bc 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ *.so *.o __pycache__/ +shipper/logshipper diff --git a/README.md b/README.md index 9f7b78e..15a889a 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,8 @@ framework) and added to `ports/king_kong_le-0.97.port`. | `event_probe.c`, `event_read.py` | the event-bus measurement instrument and its log reader — how bus event ids are identified for another game (MODE_SDK.md, "Measuring events on another game") | | `tests/test_multi_game.py` | host-compiled tests for the port selection, the ports-directory scan and the switch log (they lift the runtime's own code verbatim; the address-dependent ones need Linux — run them in the Debian container below) | | `ports/king_kong_le-0.97.port` | the King Kong LE 0.97 port: 121 verified sites, 55 named shots, 6 bus events. Installed on the machine as `/usr/local/padmode/ports/king_kong_le-0.97.port` | +| `shipper/` | `logshipper`: a small static Go service for the machine that tails `/dump/mode.log` + `/dump/switch.log` and pushes them to Loki — `go test ./shipper`, built with `shipper/build.sh`, run by `/etc/init.d/log_shipper` | +| `k8s/loki/` | the Loki the shipper pushes to: monolithic, filesystem on CephFS, 30-day retention, at `https://loki.k8s-home.lqx.net`, with the Grafana datasource auto-registered | ## Multiple games @@ -209,17 +211,57 @@ events on another game"): The ids you name that way become the `event` lines of a port (the six in `king_kong_le-0.97.port` were found this way, in the SDK's emulator). +## Log aggregation (Loki) + +The machine ships its own logs: `logshipper` (in `shipper/`) tails +`/dump/mode.log` and `/dump/switch.log` and pushes them to Loki at +`https://loki.k8s-home.lqx.net` (deployed from `k8s/loki/` on the home +cluster, Grafana datasource auto-registered). Query in Grafana or logcli: + +``` +{job="pinball", source="mode.log"} |= "ball_start" # every ball start +{job="pinball", source="mode.log"} |= "MODE START" # game modes +{job="pinball", source="switch.log"} # button edges +``` + +Labels: `job=pinball`, `machine`, `source` (`mode.log` / `switch.log`), and +`game` (from the logger's boot line, e.g. `king_kong_le`). Each line keeps its +own RFC 3339 stamp as the Loki timestamp, so live tailing in Grafana is the +game, in real time, with the machine's own clock. + +Shipper details worth knowing: + +- Per-file read offsets live in `/dump/.logshipper.state` and are saved only + after a successful push — a down Loki means retry-then-catch-up, never loss. +- First run starts at each file's **end**; run once with `SHIP_FROM_START=1` + to ship the history already on disk (Loki accepts entries up to a year old, + and the shipper calls `/flush` after a backfill — old entries are invisible + in the ingester until they reach the store). +- Truncation (`: > /dump/mode.log`) is noticed and the file restarts from 0. +- Config is env (`LOKI_URL`, `SHIP_FILES`, `SHIP_STATE`, `MACHINE`, + `SHIP_FROM_START`, `SHIP_INSECURE`, `SHIP_EVERY`); defaults match the + machine. + +Deploy the shipper: + +```bash +cd shipper && ./build.sh +scp logshipper root@stern-spike2:/tmp/ +ssh root@stern-spike2 'mount -o remount,rw /; cp /tmp/logshipper /usr/local/bin/; + cp /tmp/log_shipper /etc/init.d/ && chmod 755 /usr/local/bin/logshipper /etc/init.d/log_shipper; + ln -sf ../init.d/log_shipper /etc/rc5.d/S60log_shipper; sync; mount -o remount,ro / || true; + /etc/init.d/log_shipper start' +``` + ## Unfinished work / known issues - **The extra-ball award id is unconfirmed.** Default 13 is a static guess from the game's object-name tables; every award id is logged, so earn an extra ball, read the `AWARD` line, and pin it down (or override it in the port). -- **The master SD image does not contain the logger** — it was installed over - SSH onto the card in the machine. Making it survive a re-flash needs a - debugfs inject of `mode.so` + `game.port` + the hooked `game_monitor` into - the master image. -- **No network streaming yet.** Logs are pulled over SSH; the plan is for the - mode to ship them itself (`socket()` is not on the build's banned list). +- **The master SD image does not contain the logger or the shipper** — they + were installed over SSH onto the card in the machine. Making them survive a + re-flash needs a debugfs inject of `mode.so` + the ports + the hooked + `game_monitor` + `logshipper` into the master image. - **Cosmetic**: the boot banner still prints the upstream "score reports stay on this machine" line (see Insider Connected above). - `mode_score` is an approximation (see above). diff --git a/k8s/loki/loki.yaml b/k8s/loki/loki.yaml new file mode 100644 index 0000000..1d601da --- /dev/null +++ b/k8s/loki/loki.yaml @@ -0,0 +1,163 @@ +# Loki, monolithic: one binary, filesystem storage on a CephFS PVC, 30-day retention. +# Sized for the pinball machines' logs (a few MB a day), not for a fleet. +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: loki-config + namespace: monitoring + labels: + app: loki +data: + loki.yaml: | + auth_enabled: false + server: + http_listen_port: 3100 + grpc_listen_port: 9096 + common: + path_prefix: /var/lib/loki + replication_factor: 1 + ring: + kvstore: + store: inmemory + storage: + filesystem: + chunks_directory: /var/lib/loki/chunks + rules_directory: /var/lib/loki/rules + schema_config: + configs: + - from: 2024-01-01 + store: tsdb + object_store: filesystem + schema: v13 + index: + prefix: index_ + period: 24h + limits_config: + retention_period: 720h # 30 days + # the machines backfill their history on first ship: accept old entries + reject_old_samples: true + reject_old_samples_max_age: 8760h # a year + compactor: + working_directory: /var/lib/loki/compactor + retention_enabled: true + delete_request_store: filesystem +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: loki + namespace: monitoring + labels: + app: loki +spec: + serviceName: loki + replicas: 1 + selector: + matchLabels: + app: loki + template: + metadata: + labels: + app: loki + spec: + securityContext: + fsGroup: 10001 + runAsNonRoot: true + runAsUser: 10001 + runAsGroup: 10001 + containers: + - name: loki + image: grafana/loki:3.4.4 + args: ["-config.file=/etc/loki/loki.yaml"] + ports: + - containerPort: 3100 + name: http + resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: "1" + memory: 1Gi + readinessProbe: + httpGet: + path: /ready + port: http + initialDelaySeconds: 15 + volumeMounts: + - name: config + mountPath: /etc/loki + - name: data + mountPath: /var/lib/loki + volumes: + - name: config + configMap: + name: loki-config + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: ["ReadWriteOnce"] + storageClassName: csi-cephfs + resources: + requests: + storage: 10Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: loki + namespace: monitoring + labels: + app: loki +spec: + selector: + app: loki + ports: + - name: http + port: 3100 + targetPort: http +--- +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: loki + namespace: monitoring + annotations: + cert-manager.io/cluster-issuer: letsencrypt +spec: + ingressClassName: traefik-public + rules: + - host: loki.k8s-home.lqx.net + http: + paths: + - backend: + service: + name: loki + port: + number: 3100 + path: / + pathType: Prefix + tls: + - hosts: + - loki.k8s-home.lqx.net + secretName: loki-tls +--- +# picked up by the kube-prom Grafana's datasource sidecar +apiVersion: v1 +kind: ConfigMap +metadata: + name: grafana-loki-datasource + namespace: monitoring + labels: + grafana_datasource: "1" +data: + loki-datasource.yaml: | + apiVersion: 1 + datasources: + - name: Loki + type: loki + access: proxy + url: http://loki.monitoring.svc.cluster.local:3100 + isDefault: false diff --git a/shipper/build.sh b/shipper/build.sh new file mode 100755 index 0000000..c9bd770 --- /dev/null +++ b/shipper/build.sh @@ -0,0 +1,7 @@ +#!/bin/bash +# build.sh - cross-compile the log shipper for the Spike 2 machine (i.MX6, armv7). +# Pure Go (CGO off), so the binary is fully static and needs nothing from the machine. +set -euo pipefail +cd "$(dirname "${BASH_SOURCE[0]}")" +CGO_ENABLED=0 GOOS=linux GOARCH=arm GOARM=7 go build -trimpath -ldflags "-s -w" -o logshipper . +echo "built logshipper ($(stat -c %s logshipper 2>/dev/null || stat -f %z logshipper) bytes)" diff --git a/shipper/go.mod b/shipper/go.mod new file mode 100644 index 0000000..efe9c05 --- /dev/null +++ b/shipper/go.mod @@ -0,0 +1,3 @@ +module logshipper + +go 1.27.1 diff --git a/shipper/init.d/log_shipper b/shipper/init.d/log_shipper new file mode 100755 index 0000000..92d8880 --- /dev/null +++ b/shipper/init.d/log_shipper @@ -0,0 +1,21 @@ +#!/bin/sh +# log_shipper - tail /dump/*.log and push them to Loki (see shipper/README.md). +# Config comes from the environment; defaults in the binary match the machine. + +case "$1" in + start) + start-stop-daemon -S -b -m -p /run/logshipper.pid -x /usr/local/bin/logshipper + ;; + stop) + start-stop-daemon -K -p /run/logshipper.pid + ;; + restart) + "$0" stop + sleep 1 + "$0" start + ;; + *) + echo "usage: $0 {start|stop|restart}" + exit 1 + ;; +esac diff --git a/shipper/main.go b/shipper/main.go new file mode 100644 index 0000000..c4996af --- /dev/null +++ b/shipper/main.go @@ -0,0 +1,221 @@ +package main + +// logshipper - tails the pinball machine's gameplay logs and pushes them to Loki. +// +// Tails /dump/mode.log and /dump/switch.log, batches their lines, and POSTs them to +// Loki's /loki/api/v1/push. Per-file read offsets are saved only after a successful +// push, so a down Loki means retry-then-catch-up, never loss. Configuration is env: +// +// LOKI_URL push target (default https://loki.k8s-home.lqx.net) +// SHIP_FILES comma list (default /dump/mode.log,/dump/switch.log) +// SHIP_STATE offset state file (default /dump/.logshipper.state) +// MACHINE the machine label (default this host's name) +// SHIP_FROM_START 1 = ship from each file's start on first run (default: from its end) +// SHIP_INSECURE 1 = skip TLS verification (for a Loki without a trusted cert) +// SHIP_EVERY poll interval (default 2s) +// +// Runs under /etc/init.d/log_shipper; its stderr goes to the syslog. + +import ( + "bytes" + "crypto/tls" + "fmt" + "io" + "log" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" +) + +var ( + lokiURL = envOr("LOKI_URL", "https://loki.k8s-home.lqx.net") + files = strings.Split(envOr("SHIP_FILES", "/dump/mode.log,/dump/switch.log"), ",") + statePath = envOr("SHIP_STATE", "/dump/.logshipper.state") + machine = envOr("MACHINE", hostname()) + fromStart = os.Getenv("SHIP_FROM_START") == "1" + insecure = os.Getenv("SHIP_INSECURE") == "1" + every = 2 * time.Second + maxPush = 3 << 20 // under Loki's 4 MiB push limit +) + +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +func hostname() string { + if h, err := os.Hostname(); err == nil && h != "" { + return h + } + return "stern-spike2" +} + +// readNewLines returns the complete lines added to path since offset, and the offset to +// save once they are pushed (never past the last newline). A file truncated under us +// (": > mode.log" is safe housekeeping) restarts from 0. +func readNewLines(path string, offset int64) (lines []string, sent int64, err error) { + f, err := os.Open(path) + if err != nil { + return nil, offset, err + } + defer f.Close() + fi, err := f.Stat() + if err != nil { + return nil, offset, err + } + resume := offset + if fi.Size() < offset { + resume = 0 + } + if _, err = f.Seek(resume, io.SeekStart); err != nil { + return nil, offset, err + } + buf, err := io.ReadAll(f) + if err != nil { + return nil, resume, err + } + s := string(buf) + if cut := strings.LastIndexByte(s, '\n'); cut >= 0 { + lines = strings.Split(strings.TrimSuffix(s[:cut+1], "\n"), "\n") + sent = resume + int64(cut) + 1 + } else { + sent = resume // no complete line yet + } + return lines, sent, nil +} + +func push(payload []byte) error { + return post("/loki/api/v1/push", payload) +} + +// pushFlush asks Loki to write the ingester's chunks to the store now. +func pushFlush() error { + return post("/flush", nil) +} + +func post(path string, payload []byte) error { + transport := http.DefaultTransport.(*http.Transport).Clone() + if insecure { + transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + } + client := &http.Client{Timeout: 30 * time.Second, Transport: transport} + var body io.Reader + if payload != nil { + body = bytes.NewReader(payload) + } + resp, err := client.Post(lokiURL+path, "application/json", body) + if err != nil { + return err + } + defer resp.Body.Close() + read, _ := io.ReadAll(io.LimitReader(resp.Body, 300)) + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK { + return fmt.Errorf("loki said %s: %s", resp.Status, strings.TrimSpace(string(read))) + } + return nil +} + +type shipFile struct { + path string + source string + game string +} + +func main() { + log.SetFlags(log.LstdFlags | log.LUTC) + if v := os.Getenv("SHIP_EVERY"); v != "" { + if d, err := time.ParseDuration(v); err == nil && d > 0 { + every = d + } + } + st, err := loadState(statePath) + if err != nil { + log.Printf("state file unreadable (%v) - starting fresh", err) + st = state{} + } + var ship []shipFile + for _, path := range files { + path = strings.TrimSpace(path) + if path == "" { + continue + } + ship = append(ship, shipFile{path: path, source: baseName(path)}) + } + log.Printf("shipping %s to %s (labels: job=pinball machine=%s, every %s)", + strings.Join(files, ", "), lokiURL, machine, every) + + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + tick := time.Tick(every) + for { + for i := range ship { + f := &ship[i] + offset, known := st[baseName(f.path)] + if !known && !fromStart { + // first run: start from the file's end; run once with SHIP_FROM_START=1 + // to ship the history already on disk + if fi, err := os.Stat(f.path); err == nil { + offset = fi.Size() + } + } + lines, sent, err := readNewLines(f.path, offset) + if err != nil { + if !os.IsNotExist(err) { + log.Printf("%s: %v", f.path, err) + } + continue + } + if len(lines) == 0 { + continue + } + stamps := assignTimestamps(lines, time.Now()) + var entries []Entry + for i, line := range lines { + if g := extractGame(line); g != "" { + f.game = g + } + entries = append(entries, Entry{Source: f.source, Game: f.game, TS: stamps[i], Line: line}) + } + pushed := true + for _, payload := range buildPayloads(entries, machine, maxPush) { + if err := push(payload); err != nil { + log.Printf("push failed (%v) - %s stays at offset %d", err, f.path, offset) + pushed = false + break + } + } + if !pushed { + continue // the offset is saved only when the whole batch is in Loki + } + if needsFlush(entries, time.Now()) { + // a backfill: the old entries are invisible in the ingester's head until + // they reach the store (live tails never come through here) + if err := pushFlush(); err != nil { + log.Printf("flush failed (%v) - old entries may not be queryable yet", err) + } + } + st[baseName(f.path)] = sent + if err := saveState(statePath, st); err != nil { + log.Printf("state save failed: %v", err) + } + } + select { + case <-sig: + log.Printf("stopping") + return + case <-tick: + } + } +} + +func baseName(path string) string { + if i := strings.LastIndexByte(path, '/'); i >= 0 { + return path[i+1:] + } + return path +} diff --git a/shipper/shipper.go b/shipper/shipper.go new file mode 100644 index 0000000..dafbd8e --- /dev/null +++ b/shipper/shipper.go @@ -0,0 +1,230 @@ +package main + +// The pure half of the shipper: line parsing, the Loki push payload, the offset state. +// main.go is the tail-and-push loop around it. + +import ( + "encoding/json" + "os" + "regexp" + "strconv" + "time" +) + +const stampLayout = "2006-01-02T15:04:05.000Z" + +var stampRe = regexp.MustCompile(`\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z`) +var gameRe = regexp.MustCompile(`up: game (\S+)`) + +// extractTimestamp finds the line's own RFC 3339 (millisecond, UTC) stamp wherever it sits: +// the runtime prefixes an uptime count and a [who], gamelog puts its stamp after its tag, +// the switch log starts with it. A line without a stamp (the boot banner) reports false and +// the caller stamps it with the shipping time. +func extractTimestamp(line string) (time.Time, bool) { + s := stampRe.FindString(line) + if s == "" { + return time.Time{}, false + } + ts, err := time.Parse(stampLayout, s) + if err != nil { + return time.Time{}, false + } + return ts, true +} + +// extractGame names the game from the boot line ("up: game king_kong_le 0.97, ..."), "" elsewhere. +func extractGame(line string) string { + m := gameRe.FindStringSubmatch(line) + if m == nil { + return "" + } + return m[1] +} + +// assignTimestamps gives every line a timestamp: its own stamp where it has one; an +// unstamped line (the boot banner) takes the NEXT stamped line's time — a stream must +// not start at the shipping time and then jump back to the game's boot — and a trailing +// unstamped line takes the previous stamped time. fallback covers a batch with no +// stamps at all. +func assignTimestamps(lines []string, fallback time.Time) []time.Time { + stamps := make([]time.Time, len(lines)) + next := time.Time{} + for i := len(lines) - 1; i >= 0; i-- { + if ts, ok := extractTimestamp(lines[i]); ok { + next = ts + } + stamps[i] = next + } + prev := time.Time{} + for i := range lines { + if ts, ok := extractTimestamp(lines[i]); ok { + prev = ts + } + if stamps[i].IsZero() { + stamps[i] = prev + } + } + for i := range stamps { + if stamps[i].IsZero() { + stamps[i] = fallback + } + } + return stamps +} + +// Entry is one shipped line. +type Entry struct { + Source string + Game string + TS time.Time + Line string +} + +// lokiPush is the /loki/api/v1/push body: labelled streams of (nanoseconds, line) pairs. +type lokiPush struct { + Streams []lokiStream `json:"streams"` +} + +type lokiStream struct { + Stream map[string]string `json:"stream"` + Values [][]string `json:"values"` +} + +// buildPayloads groups entries by their label set and marshals Loki push bodies, each JSON +// under maxBytes (Loki refuses a push past its limit, so a big batch is split). +func buildPayloads(entries []Entry, machine string, maxBytes int) [][]byte { + type group struct { + labels map[string]string + values [][]string + } + groups := map[string]*group{} + var order []string + for _, e := range entries { + ts, ok := extractTimestamp(e.Line) + if !ok { + ts = e.TS + } + labels := map[string]string{"job": "pinball", "machine": machine, "source": e.Source} + if e.Game != "" { + labels["game"] = e.Game + } + key := e.Source + "\x00" + e.Game + g, ok := groups[key] + if !ok { + groups[key] = &group{labels: labels} + g = groups[key] + order = append(order, key) + } + g.values = append(g.values, []string{strconv.FormatInt(ts.UnixNano(), 10), e.Line}) + } + + var payloads [][]byte + var cur *lokiPush + curBytes := 0 + flush := func() { + if cur != nil && len(cur.Streams) > 0 { + if b, err := json.Marshal(cur); err == nil { + payloads = append(payloads, b) + } + } + cur = nil + curBytes = 0 + } + add := func(s lokiStream) int { + p := lokiPush{Streams: []lokiStream{s}} + b, err := json.Marshal(p) + if err != nil { + return 0 + } + if cur == nil { + cur = &lokiPush{} + } + cur.Streams = append(cur.Streams, s) + curBytes += len(b) + return len(b) + } + for _, key := range order { + g := groups[key] + for len(g.values) > 0 { + // a stream that fits whole goes into the current payload; one that does not is + // halved until a piece fits, then the payload is flushed and the rest continues + take := g.values + var piece []lokiStream + for { + s := lokiStream{Stream: g.labels, Values: take} + p := lokiPush{Streams: []lokiStream{s}} + b, err := json.Marshal(p) + if err != nil { + break + } + if len(b)+curBytes <= maxBytes { + piece = []lokiStream{s} + break + } + if len(take) == 1 { + // a single line over the cap is shipped alone anyway (it cannot split) + piece = []lokiStream{s} + break + } + take = take[:len(take)/2] + } + if piece == nil { + flush() + continue + } + add(piece[0]) + g.values = g.values[len(piece[0].Values):] + if len(g.values) > 0 || curBytes > maxBytes/2 { + flush() + } + } + } + flush() + return payloads +} + +// needsFlush: entries older than the out-of-order window (Loki serves head entries only +// within max_chunk_age/2 ≈ 1h) sit invisible in the ingester until a flush puts them in the +// store. A backfill (first ship of history) needs one; a live tail never does. +func needsFlush(entries []Entry, now time.Time) bool { + for _, e := range entries { + if now.Sub(e.TS) > time.Hour { + return true + } + } + return false +} + +// ---- the offset state ------------------------------------------------------------------- + +type state map[string]int64 + +// loadState reads the saved per-file offsets; a missing file is a fresh start, not an error. +func loadState(path string) (state, error) { + b, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return state{}, nil + } + return state{}, err + } + var st state + if err := json.Unmarshal(b, &st); err != nil { + return state{}, err + } + return st, nil +} + +// saveState writes the offsets atomically (a tmp file and a rename), so a crash mid-write +// keeps the last fully saved state. +func saveState(path string, st state) error { + b, err := json.Marshal(st) + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, b, 0644); err != nil { + return err + } + return os.Rename(tmp, path) +} diff --git a/shipper/shipper_test.go b/shipper/shipper_test.go new file mode 100644 index 0000000..64114bb --- /dev/null +++ b/shipper/shipper_test.go @@ -0,0 +1,191 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +// ---- timestamps ----------------------------------------------------------------------- + +func TestExtractsTheRfc3339StampFromEveryLineShape(t *testing.T) { + // the runtime writes an uptime-ms prefix before [who]; gamelog puts its stamp after the tag; + // the switch log starts with the stamp. All three must parse, wherever the stamp sits. + lines := []string{ + ` 1234 [pad] [GAMELOG] 2026-10-12T20:45:12.123Z EV ball_start player=1 ball=1 score=0`, + ` 1300 [pad] gamelog: 2026-10-12T20:45:12.500Z MODE START 5 (Island Scene) player=1`, + `2026-10-12T20:45:12.678Z SW 62 "RIGHT FLIPPER BUTTON" level=1 hit=1 game=1`, + ` 1400 [pad] port /usr/local/padmode/ports/king_kong_le-0.97.port: king_kong_le 0.97, 121 sites, 55 shots`, + } + want := []bool{true, true, true, false} + wanted := []time.Time{ + time.Date(2026, 10, 12, 20, 45, 12, 123000000, time.UTC), + time.Date(2026, 10, 12, 20, 45, 12, 500000000, time.UTC), + time.Date(2026, 10, 12, 20, 45, 12, 678000000, time.UTC), + } + for i, line := range lines { + ts, ok := extractTimestamp(line) + if ok != want[i] { + t.Fatalf("line %d: ok=%v, want %v", i, ok, want[i]) + } + if ok && !ts.Equal(wanted[i]) { + t.Fatalf("line %d: wrong stamp %v", i, ts) + } + } +} + +// ---- the game label -------------------------------------------------------------------- + +func TestExtractsTheGameFromTheBootLine(t *testing.T) { + for _, line := range []string{ + ` 1234 [pad] [GAMELOG] 2026-10-12T20:45:12.123Z up: game king_kong_le 0.97, 55 named shots; event ids 77/37/52/40/200/198; extra_ball_award=13`, + `[GAMELOG] 2026-10-13T09:00:00.000Z up: game godzilla_le 1.16, 40 named shots`, + } { + if g := extractGame(line); g != "king_kong_le" && g != "godzilla_le" { + t.Fatalf("extractGame(%q) = %q", line, g) + } + } + if g := extractGame(` 1234 [pad] [GAMELOG] 2026-10-12T20:45:12.123Z EV bonus_end`); g != "" { + t.Fatalf("a non-boot line named a game: %q", g) + } +} + +// ---- the Loki payload ------------------------------------------------------------------- + +func TestPayloadGroupsByLabelsAndCarriesNanosecondStamps(t *testing.T) { + ts := time.Date(2026, 10, 12, 20, 45, 12, 123000000, time.UTC) + entries := []Entry{ + {Source: "mode", Game: "king_kong_le", TS: ts, Line: "one"}, + {Source: "mode", Game: "king_kong_le", TS: ts.Add(time.Second), Line: "two"}, + {Source: "switch", Game: "", TS: ts, Line: "three"}, + } + payloads := buildPayloads(entries, "stern-spike2", 1<<20) + if len(payloads) != 1 { + t.Fatalf("got %d payloads, want 1 (both sources fit under the cap)", len(payloads)) + } + var parsed struct { + Streams []struct { + Stream map[string]string `json:"stream"` + Values [][]string `json:"values"` + } `json:"streams"` + } + if err := json.Unmarshal(payloads[0], &parsed); err != nil { + t.Fatalf("payload is not JSON: %v", err) + } + if len(parsed.Streams) != 2 { + t.Fatalf("got %d streams, want 2 (mode and switch have different labels)", len(parsed.Streams)) + } + for _, s := range parsed.Streams { + if s.Stream["job"] != "pinball" || s.Stream["machine"] != "stern-spike2" { + t.Fatalf("bad labels: %v", s.Stream) + } + } + // find the mode stream and check ordering, stamps and the game label + var mode struct { + values [][]string + } + for _, s := range parsed.Streams { + if s.Stream["source"] == "mode" { + if s.Stream["game"] != "king_kong_le" { + t.Fatalf("mode stream game label = %q", s.Stream["game"]) + } + mode.values = s.Values + } + } + if len(mode.values) != 2 || mode.values[0][1] != "one" || mode.values[1][1] != "two" { + t.Fatalf("mode values wrong: %v", mode.values) + } + if mode.values[0][0] != "1791837912123000000" { // the line's own stamp, nanoseconds, as a string + t.Fatalf("stamp = %q", mode.values[0][0]) + } +} + +func TestPayloadSplitsPastTheCap(t *testing.T) { + ts := time.Date(2026, 10, 12, 20, 45, 12, 0, time.UTC) + var entries []Entry + for i := 0; i < 100; i++ { + entries = append(entries, Entry{Source: "mode", TS: ts, Line: strings.Repeat("x", 1000)}) + } + payloads := buildPayloads(entries, "stern-spike2", 4096) + if len(payloads) < 2 { + t.Fatalf("got %d payloads, want several", len(payloads)) + } + for _, p := range payloads { + if len(p) > 4096 { + t.Fatalf("payload of %d bytes exceeds the cap", len(p)) + } + } +} + +// ---- stamps for unstamped lines --------------------------------------------------------- + +func TestUnstampedLinesInheritTheirNeighboursStamps(t *testing.T) { + stamp := func(s string) time.Time { + ts, _ := time.Parse(stampLayout, s) + return ts + } + fallback := stamp("2026-10-22T12:00:00.000Z") + lines := []string{ + ` 1234 [pad] port /usr/local/padmode/ports/king_kong_le-0.97.port: king_kong_le 0.97, 121 sites, 55 shots`, + ` 1234 [pad] [GAMELOG] 2026-10-12T20:45:12.123Z up: game king_kong_le 0.97, 55 named shots`, + ` 1250 [pad] [GAMELOG] 2026-10-12T20:45:12.500Z EV game_start - GAME START`, + ` 1300 [pad] [GAMELOG] 2026-10-12T20:45:13.000Z EV ball_start player=1 ball=1 score=0`, + ` 1400 [pad] gamelog: no stamp on this one`, + } + got := assignTimestamps(lines, fallback) + // the banner comes BEFORE the first stamp: it takes the next line's stamp, not the + // fallback - a stream must not start at "now" and then jump ten days back + if !got[0].Equal(stamp("2026-10-12T20:45:12.123Z")) { + t.Fatalf("banner stamp = %v, want the next line's stamp", got[0]) + } + for i, want := range []string{"2026-10-12T20:45:12.123Z", "2026-10-12T20:45:12.123Z", + "2026-10-12T20:45:12.500Z", "2026-10-12T20:45:13.000Z", "2026-10-12T20:45:13.000Z"} { + if !got[i].Equal(stamp(want)) { + t.Fatalf("line %d stamp = %v, want %v", i, got[i], want) + } + } + + // a batch with no stamps at all falls back to the shipping time + got = assignTimestamps([]string{"one", "two"}, fallback) + if !got[0].Equal(fallback) || !got[1].Equal(fallback) { + t.Fatalf("fallback stamps wrong: %v", got) + } +} + +// ---- the flush after a backfill --------------------------------------------------------- + +func TestOldEntriesNeedAFlush(t *testing.T) { + // entries older than the out-of-order window (1h) sit invisible in the ingester's head + // until a flush; live entries never need one + now := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC) + fresh := []Entry{{Source: "mode", TS: now.Add(-time.Minute), Line: "x"}} + old := []Entry{{Source: "mode", TS: now.Add(-10 * 24 * time.Hour), Line: "x"}} + if needsFlush(fresh, now) { + t.Fatal("a fresh batch must not trigger a flush") + } + if !needsFlush(old, now) { + t.Fatal("an old batch must trigger a flush") + } +} + +// ---- the offset state ------------------------------------------------------------------- + +func TestStateRoundTrips(t *testing.T) { + dir := t.TempDir() + path := dir + "/state" + st := state{"mode.log": 12345, "switch.log": 67890} + if err := saveState(path, st); err != nil { + t.Fatalf("save: %v", err) + } + got, err := loadState(path) + if err != nil { + t.Fatalf("load: %v", err) + } + if got["mode.log"] != 12345 || got["switch.log"] != 67890 { + t.Fatalf("round trip lost offsets: %v", got) + } + if _, err := loadState(path + "-missing"); err != nil { + t.Fatalf("a missing state file is not an error: %v", err) + } +}