Case study: auditing structured application logs

Copy Markdown View Source

Every other example in EXAMPLES.md is deliberately small — one feature, one paragraph, one screenful. This one is deliberately not: a single, complete, real-world-shaped program that walks through why you'd reach for Episteme, not just how to call one predicate. It assumes you're comfortable with the vocabulary from TUTORIAL.md and pulls in a feature from nearly every corner of the library — a DCG grammar, the dynamic database, aggregation with existential quantification, and format/2 — to solve one coherent problem rather than eight disconnected ones.

0. Shared setup

alias Episteme.{Database, Term}
alias Episteme.Term.Compound
c = fn name, args -> %Compound{name: name, args: args} end
embed = fn goal -> %Compound{name: :{}, args: [goal]} end

c is the same shorthand EXAMPLES.md §0 uses. embed is new here: it wraps an ordinary goal as {Goal} — ISO's own {}/1 term shape — the way you drop a plain predicate call into the middle of a DCG body without it being treated as grammar.

Intent

A lot of real services — Heroku's own routing layer, most Go and Rust web frameworks, plenty of hand-rolled ones — emit logs in logfmt: one line per event, space-separated key=value pairs, no fixed schema beyond a handful of conventional fields. A line looks like this:

ts=2026-08-05T10:15:32Z level=warn component=auth event=login_failed user=alice ip=203.0.113.5

Say you want a small audit tool over a stream of these: flag any user with repeated failed logins (a brute-force indicator), and separately flag any IP address behind failed logins for more than one distinct user (a credential-stuffing indicator — one attacker, many stolen usernames, one source). Both are real heuristics real fraud/security tooling actually uses, and both are naturally relational questions — "group these facts by X, then check something about each group" — which is exactly the shape logic programming is for.

Three things about Episteme fit this well:

  • No parser of its own, but a DCG when you need one. Episteme doesn't ship a logfmt reader — nothing does, out of the box — but DCG support lets you write the grammar as a grammar, not a hand-rolled loop of string splits.
  • A mutable database for a line-at-a-time stream. Log lines arrive one at a time, not as a batch you already have in memory; the dynamic database (assertz/1) is built for exactly that.
  • Aggregation that groups for you. "Which users have ≥ 3 failed logins" and "which IPs are shared across users" are both grouping queries — setof/3's free-variable grouping does the work a hand-written accumulator would otherwise need.

Solution

Parsing: a DCG grammar for logfmt lines

Tokenizing on whitespace is just String.split/1 — Elixir already does that well, so there's no reason to reinvent it inside Episteme. What the grammar below owns is everything after that: recognizing the shape of a line (four required fields in order, then a variable-length tail of arbitrary ones) and pulling each key=value token apart.

any_field/2 consumes exactly one token and splits it around =:

af_key = Term.new_var("Key")
af_value = Term.new_var("Value")
af_token = Term.new_var("Token")
af_key_str = Term.new_var("KeyStr")

any_field_head = c.(:any_field, [af_key, af_value])

any_field_body =
  c.(:and, [
    [af_token],
    embed.(
      c.(:and, [
        c.(:split_string, [af_token, "=", "", [af_key_str, af_value]]),
        c.(:atom_string, [af_key, af_key_str])
      ])
    )
  ])

db = Database.add_clause(Database.new(), Episteme.Dcg.translate_rule(any_field_head, any_field_body))

Read the DCG body the way Episteme.Dcg reads it: [af_token] is a one-element terminal — "consume the next token from the input and call it af_token" (a terminal doesn't have to be a literal; a variable inside one just captures whatever's there). The embed.(...) half is an ordinary goal, spliced in without touching the input list: split_string/4 breaks "level=warn" into ["level", "warn"], and atom_string/2 turns "level" into the atom :level so field names can be matched and grouped on later. Database.add_clause/2 plus Episteme.Dcg.translate_rule/2 (rather than consult_forms/2's {:dcg, ...} shorthand) is the direct call the shorthand itself builds on, and reads consistently with how every other rule in this file gets stored.

fields/1 is the variable-length tail: zero or more any_field/2 matches, collected into a list.

db = Database.add_clause(db, Episteme.Dcg.translate_rule(c.(:fields, [[]]), []))

fc_k = Term.new_var("K")
fc_v = Term.new_var("V")
fc_rest = Term.new_var("Rest")

fields_cons_head = c.(:fields, [[c.(:field, [fc_k, fc_v]) | fc_rest]])
fields_cons_body = c.(:and, [c.(:any_field, [fc_k, fc_v]), c.(:fields, [fc_rest])])

db = Database.add_clause(db, Episteme.Dcg.translate_rule(fields_cons_head, fields_cons_body))

Same recursive shape as sum_list/2 in EXAMPLES.md §7: an empty-list base case, and a case that peels one field(K, V) off the front and recurses on the rest — except here the recursion is consuming tokens, not walking an already-built list.

log_line/5 ties it together: four required fields in a fixed order, each converted to an atom, then whatever's left goes through fields/1:

ll_ts = Term.new_var("Ts")
ll_level = Term.new_var("Level")
ll_level_str = Term.new_var("LevelStr")
ll_component = Term.new_var("Component")
ll_component_str = Term.new_var("ComponentStr")
ll_event = Term.new_var("Event")
ll_event_str = Term.new_var("EventStr")
ll_fields = Term.new_var("Fields")

log_line_head = c.(:log_line, [ll_ts, ll_level, ll_component, ll_event, ll_fields])

log_line_body =
  c.(:and, [
    c.(:any_field, [:ts, ll_ts]),
    c.(:and, [
      c.(:any_field, [:level, ll_level_str]),
      c.(:and, [
        embed.(c.(:atom_string, [ll_level, ll_level_str])),
        c.(:and, [
          c.(:any_field, [:component, ll_component_str]),
          c.(:and, [
            embed.(c.(:atom_string, [ll_component, ll_component_str])),
            c.(:and, [
              c.(:any_field, [:event, ll_event_str]),
              c.(:and, [
                embed.(c.(:atom_string, [ll_event, ll_event_str])),
                c.(:fields, [ll_fields])
              ])
            ])
          ])
        ])
      ])
    ])
  ])

db = Database.add_clause(db, Episteme.Dcg.translate_rule(log_line_head, log_line_body))

any_field(:ts, Ts) deliberately leaves the timestamp as a string rather than converting it to an atom like the other three — ISO 8601 timestamps sort correctly as plain byte strings, so there's no need to parse them any further for what this audit actually asks. Trying the grammar directly with phrase/2:

line = "ts=2026-08-05T10:15:32Z level=warn component=auth event=login_failed user=alice ip=203.0.113.5"

parse_goal =
  c.(:phrase, [
    c.(:log_line, [
      Term.new_var("Ts"),
      Term.new_var("Level"),
      Term.new_var("Component"),
      Term.new_var("Event"),
      Term.new_var("Fields")
    ]),
    String.split(line)
  ])

Episteme.query(parse_goal, db)
#=> {:ok, [%{
#     "Ts" => "2026-08-05T10:15:32Z",
#     "Level" => :warn,
#     "Component" => :auth,
#     "Event" => :login_failed,
#     "Fields" => [
#       %Episteme.Term.Compound{name: :field, args: [:user, "alice"]},
#       %Episteme.Term.Compound{name: :field, args: [:ip, "203.0.113.5"]}
#     ]
#   }]}

Rules over the parsed facts

Every line becomes a log_entry(Ts, Level, Component, Event, Fields) fact once parsed. field_value/3 is a lookup into that trailing Fields list, built directly on the member/2 builtin:

fv_fields = Term.new_var("Fields")
fv_key = Term.new_var("Key")
fv_value = Term.new_var("Value")

db =
  Database.add_clause(
    db,
    {c.(:field_value, [fv_fields, fv_key, fv_value]),
     c.(:member, [c.(:field, [fv_key, fv_value]), fv_fields])}
  )

failed_login/3 picks out the one event shape the rest of this cares about:

fl_user = Term.new_var("User")
fl_ip = Term.new_var("Ip")
fl_ts = Term.new_var("Ts")
fl_fields = Term.new_var("Fields")

db =
  Database.add_clause(
    db,
    {c.(:failed_login, [fl_user, fl_ip, fl_ts]),
     c.(:and, [
       c.(:log_entry, [fl_ts, :warn, :auth, :login_failed, fl_fields]),
       c.(:and, [
         c.(:field_value, [fl_fields, :user, fl_user]),
         c.(:field_value, [fl_fields, :ip, fl_ip])
       ])
     ])}
  )

Now the two heuristics, each a setof/3 grouped by exactly one of failed_login/3's three fields, with the other free variable folded away via ^ (EXAMPLES.md explains the Var^Goal existential-quantification syntax in full). brute_force_suspect/1 groups by User, folding Ip away, and checks the group is big enough:

bf_user = Term.new_var("User")
bf_ip = Term.new_var()
bf_ts = Term.new_var()
bf_timestamps = Term.new_var("Timestamps")
bf_n = Term.new_var("N")

db =
  Database.add_clause(
    db,
    {c.(:brute_force_suspect, [bf_user]),
     c.(:and, [
       c.(:setof, [bf_ts, c.(:^, [bf_ip, c.(:failed_login, [bf_user, bf_ip, bf_ts])]), bf_timestamps]),
       c.(:and, [c.(:length, [bf_timestamps, bf_n]), c.(:greater_or_equal, [bf_n, 3])])
     ])}
  )

shared_ip_suspect/1 is the mirror image — group by Ip, fold Ts away, check more than one distinct user shows up:

si_ip = Term.new_var("Ip")
si_user = Term.new_var("User")
si_ts = Term.new_var()
si_users = Term.new_var("Users")
si_n = Term.new_var("N")

db =
  Database.add_clause(
    db,
    {c.(:shared_ip_suspect, [si_ip]),
     c.(:and, [
       c.(:setof, [si_user, c.(:^, [si_ts, c.(:failed_login, [si_user, si_ip, si_ts])]), si_users]),
       c.(:and, [c.(:length, [si_users, si_n]), c.(:>, [si_n, 1])])
     ])}
  )

(greater_or_equal/2 and >/2 — see Arithmetic comparisons — setof/3 already deduplicates si_users/bf_timestamps, so length/2 here is counting distinct values, not raw match counts.)

Ingesting a batch of log lines

This is the dynamic-database half: parse each raw line and assertz the resulting fact, the same assertz/1 from EXAMPLES.md §4, one line at a time — exactly the shape a real log stream arrives in.

lines = [
  "ts=2026-08-05T10:15:32Z level=warn component=auth event=login_failed user=alice ip=203.0.113.5",
  "ts=2026-08-05T10:15:40Z level=info component=auth event=login_success user=bob ip=198.51.100.7",
  "ts=2026-08-05T10:16:02Z level=warn component=auth event=login_failed user=alice ip=203.0.113.5",
  "ts=2026-08-05T10:16:10Z level=error component=payments event=charge_failed user=carol amount=49.99",
  "ts=2026-08-05T10:16:45Z level=warn component=auth event=login_failed user=alice ip=203.0.113.9",
  "ts=2026-08-05T10:16:51Z level=warn component=auth event=login_failed user=dave ip=203.0.113.9",
  "ts=2026-08-05T10:17:03Z level=info component=auth event=login_success user=alice ip=203.0.113.9"
]

Enum.each(lines, fn line ->
  goal =
    c.(:phrase, [
      c.(:log_line, [
        Term.new_var("Ts"),
        Term.new_var("Level"),
        Term.new_var("Component"),
        Term.new_var("Event"),
        Term.new_var("Fields")
      ]),
      String.split(line)
    ])

  {:ok, [solution]} = Episteme.query(goal, db)

  entry =
    c.(:log_entry, [
      solution["Ts"],
      solution["Level"],
      solution["Component"],
      solution["Event"],
      solution["Fields"]
    ])

  {:ok, [%{}]} = Episteme.query(c.(:assertz, [entry]), db)
end)

Note carol's line: a payments/charge_failed event with an amount field instead of ip at all. log_entry/5's trailing Fields list is exactly why that's not a problem — the grammar never hard-codes which fields follow the first four, so entries with completely different shapes live in the same database without any schema migration.

Running the audit

report_user = Term.new_var()
report_ip = Term.new_var()
report_suspects = Term.new_var("Suspects")
report_ips = Term.new_var("SharedIps")

report_goal =
  c.(:and, [
    c.(:findall, [report_user, c.(:brute_force_suspect, [report_user]), report_suspects]),
    c.(:and, [
      c.(:findall, [report_ip, c.(:shared_ip_suspect, [report_ip]), report_ips]),
      c.(:and, [
        c.(:format, ["Brute-force suspects: ~w~n", [report_suspects]]),
        c.(:format, ["Shared-IP suspects: ~w~n", [report_ips]])
      ])
    ])
  ])

Episteme.query(report_goal, db)

Result

Actually running the above (mix run against a checkout of this version of Episteme) prints:

Brute-force suspects: ["alice"]
Shared-IP suspects: ["203.0.113.9"]

and answers {:ok, [%{"Suspects" => ["alice"], "SharedIps" => ["203.0.113.9"]}]}.

Both are correct, and for different reasons — which is the point of running two independent heuristics instead of one:

  • alice is a brute-force suspect: three failed logins in the data, full stop. It doesn't matter that two of them came from one IP (203.0.113.5) and the third from another (203.0.113.9) — brute_force_suspect/1 explicitly folds Ip away with ^ before counting, because "did this account get hammered" shouldn't depend on the attacker rotating source addresses.
  • 203.0.113.9 is a shared-IP suspect: it shows up behind failed logins for two different users (alice and dave), which is exactly the "one attacker, many stolen credentials" shape — independent of the fact that dave alone only has one failed login and would never trip the brute-force check by himself.
  • bob never appears in either report — his only log line is a login_success, which failed_login/3 doesn't match at all.
  • carol doesn't appear either, for a different reason: her line is a real, successfully-parsed log_entry/5 fact (the DCG doesn't care that it's a payment event, not a login), it's just that failed_login/3 only ever looks at warn/auth/login_failed entries, so a charge_failed payment event was never in scope for either audit to begin with.

That last distinction — parsed but not relevant vs. never matched the grammar at all — is worth sitting with: nothing here had to special-case carol's line, reject it, or fail. It's a perfectly ordinary fact in the database; it simply never satisfies a goal that specifically asks about failed logins. A schema-first approach (a fixed struct per log line, one JSON schema, etc.) would have had to decide up front whether payment events belong in the same table as auth events; here that question just never comes up.

What this exercises

Every one of these is a full REFERENCE.md entry if you want the complete semantics: