Recipe: Ingest stdin Logs

Use stdin input for shell pipelines, terminal output, or process logs that already produce one JSON object per line.

Pipe logs into the agent

journalctl -u checkout -f -o json | \
  blazerules_agent \
    --rules rules.yaml \
    --input stdin \
    --batch-size 2048 \
    --flush-ms 250 \
    --output stdout \
    --service checkout \
    --source journald

Python equivalent: evaluate stdin as one batch

import sys
import blazerules

engine = blazerules.RuleEngine()
engine.load_rules("rules.yaml")

payload = sys.stdin.buffer.read()
result = engine.evaluate_ndjson(payload)
print(result.grouped_decision_indices())

For a file-backed decision log:

tail -F app.ndjson | \
  blazerules_agent \
    --rules rules.yaml \
    --input stdin \
    --output ndjson \
    --output-path decisions.ndjson

Python equivalent: launch the same stdin agent

import subprocess

tail = subprocess.Popen(["tail", "-F", "app.ndjson"], stdout=subprocess.PIPE)
agent = subprocess.Popen([
    "blazerules_agent",
    "--rules", "rules.yaml",
    "--input", "stdin",
    "--output", "ndjson",
    "--output-path", "decisions.ndjson",
], stdin=tail.stdout)
agent.wait()

Input Contract

stdin input expects NDJSON:

{"event_id":"e1","message":"checkout started","amount":30}
{"event_id":"e2","message":"checkout failed","amount":900}

Each line becomes one record. The agent buffers lines and evaluates batches by batch_size or flush_ms.

Duplicate Events

Use repeated --dedupe-key flags when the upstream log source may replay lines:

cat events.ndjson | \
  blazerules_agent \
    --rules rules.yaml \
    --input stdin \
    --output stdout \
    --dedupe-key event_id \
    --dedupe-ttl-seconds 3600

Python equivalent: run the same de-dupe agent

import subprocess

with open("events.ndjson", "rb") as src:
    subprocess.run([
        "blazerules_agent",
        "--rules", "rules.yaml",
        "--input", "stdin",
        "--output", "stdout",
        "--dedupe-key", "event_id",
        "--dedupe-ttl-seconds", "3600",
    ], stdin=src, check=True)

The duplicate window is local to the agent process.