feat: log aggregation — the machine ships its logs to Loki
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 <noreply@anthropic.com>
This commit is contained in:
Claude Code
@@ -2,3 +2,4 @@
|
||||
*.so
|
||||
*.o
|
||||
__pycache__/
|
||||
shipper/logshipper
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
Executable
+7
@@ -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)"
|
||||
@@ -0,0 +1,3 @@
|
||||
module logshipper
|
||||
|
||||
go 1.27.1
|
||||
Executable
+21
@@ -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
|
||||
+221
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user