Building a Home SIEM, Part 2: Connecting Log Sources

At the end of Part 1, we had a working Elastic Stack and absolutely nothing in it. The plumbing worked; no water was flowing.

This part fixes that. We'll connect three real log sources — Pi-hole (DNS), Traefik (reverse proxy) and Authentik (SSO) — write a Logstash pipeline that routes and parses each one into its own index, and run our first searches in Kibana.

This is where it gets genuinely satisfying, and also where most of my snags lived. Real-world logs are messy, every source ships them differently, and almost nothing worked the way the documentation implied on the first try. As before, I've left the gotchas in.

The sources, and how each one ships

Everything lands on the same Logstash syslog listener we opened in Part 1 (port 5514), but each source gets there differently:

Why not just use the Docker syslog driver everywhere? Because it creates a runtime dependency on the SIEM. With the syslog driver, a container's log output goes straight to Logstash over TCP. If Logstash is down when that container starts, the container can block or hang.

That's tolerable for a non-critical service. It's a terrible idea for your reverse proxy. If Traefik hangs waiting for Logstash, everything behind it goes down — including Kibana, the tool you'd use to diagnose the problem.

So my rule became:

• Critical infrastructure (Traefik) → write logs to local files, ship them with a separate script. The service never depends on the SIEM being alive.
• Non-critical services (Authentik) → the syslog driver is acceptable, as long as you know the trade-off.

I learned how real that trade-off is the hard way. More on that in the gotchas.

The Logstash pipeline

I'm showing the whole pipeline up front so you can see the shape, then walking through each source. You don't need all of it on day one — start with the Pi-hole branches and add the others as you go.

The idea is simple: one syslog input, then route every event by its syslog tag (which Logstash puts in [process][name]), parse it the right way for that source, and send it to its own index.

On the Elastic VM:

nano ~/siem/logstash/pipeline/siem.conf

input {
  syslog {
    port => 5514
    type => "syslog"
  }
}

filter {

  # ---- Authentik (structured JSON) ----
  if [process][name] == "authentik" {
    mutate { replace => { "type" => "authentik" } }
    json {
      source => "message"
      target => "authentik"
    }
    date {
      match => ["[authentik][timestamp]", "ISO8601"]
      target => "@timestamp"
    }
  }

  # ---- Traefik access log (structured JSON) ----
  else if [process][name] == "traefik" {
    mutate { replace => { "type" => "traefik" } }
    json {
      source => "message"
      target => "traefik"
    }
    date {
      match => ["[traefik][time]", "ISO8601"]
      target => "@timestamp"
    }
  }

  # ---- Traefik application log (structured JSON) ----
  else if [process][name] == "traefik-app" {
    mutate { replace => { "type" => "traefik" } }
    mutate { add_tag => ["traefik_app"] }
    json {
      source => "message"
      target => "traefik"
    }
    date {
      match => ["[traefik][time]", "ISO8601"]
      target => "@timestamp"
    }
  }

  # ---- Pi-hole blocked queries (from the SQLite poller) ----
  else if [process][name] == "pihole-blocked" {
    mutate { replace => { "type" => "pihole" } }
    mutate { add_tag => ["blocked"] }
    grok {
      match => {
        "message" => "id=%{NUMBER:query_id} domain=%{HOSTNAME:query_domain} client=%{IP:client_ip} status=%{NUMBER:block_status}"
      }
    }
  }

  # ---- Pi-hole DNS queries (from pihole.log) ----
  else if [process][name] == "pihole" {
    mutate { replace => { "type" => "pihole" } }
    grok {
      match => {
        "message" => "%{SYSLOGTIMESTAMP:log_timestamp} dnsmasq\[%{NUMBER:pid}\]: %{GREEDYDATA:dns_message}"
      }
    }
    date {
      match => ["log_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
      timezone => "Europe/Dublin"
    }
    if "query[" in [dns_message] {
      mutate { add_tag => ["dns_query"] }
      grok {
        match => {
          "dns_message" => "query\[%{WORD:query_type}\] %{HOSTNAME:query_domain} from %{IP:client_ip}"
        }
      }
    }
    if "forwarded" in [dns_message] { mutate { add_tag => ["forwarded"] } }
    if "cached" in [dns_message]    { mutate { add_tag => ["cached"] } }
  }
}

output {
  if [type] == "authentik" {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      user => "elastic"
      password => "${ELASTIC_PASSWORD}"
      index => "authentik-%{+YYYY.MM.dd}"
    }
  }
  else if [type] == "traefik" {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      user => "elastic"
      password => "${ELASTIC_PASSWORD}"
      index => "traefik-%{+YYYY.MM.dd}"
    }
  }
  else {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      user => "elastic"
      password => "${ELASTIC_PASSWORD}"
      index => "pihole-dns-%{+YYYY.MM.dd}"
    }
  }
}

Set timezone to your own if you're not in Ireland. Then restart Logstash and watch it start:

cd ~/siem
sudo docker compose restart logstash
sudo docker compose logs logstash --tail 30

You're looking for the pipeline starting and syslog listeners on TCP and UDP 5514. Three things in this file cost me real time, so read them before you edit it:

• [process][name], not [process.name]. The syslog input stores fields as nested objects. The dotted form looks right, matches nothing, and fails silently — every event falls through to the last branch.
• Filters go in filter, outputs go in output. Paste a mutate block into the output section by mistake and Logstash refuses to start with Couldn't find any output plugin named 'mutate'. Count your braces.
• The file must end in .conf. My first version was pihole.config. Logstash ignored it completely and I spent an evening wondering why nothing was parsed.

Source 1: Pi-hole DNS

DNS is the best first source. Almost everything on your network does a DNS lookup before it does anything else, so it's a near-complete record of what's talking to what.

My original plan was the obvious one: point rsyslog on the Pi at Logstash. It worked for about ten minutes. Pi-hole v6's FTL engine doesn't reliably route through syslog, so the logs dried up. A small script that tails Pi-hole's log file proved far more dependable.

All of this runs on the Pi-hole itself. First, make sure netcat is installed:

sudo apt install netcat-openbsd -y

Script 1: DNS queries

sudo nano /usr/local/bin/pihole-siem.sh

#!/bin/bash
# Forward Pi-hole DNS query logs to Logstash

LOGSTASH_HOST="YOUR_ELASTIC_VM_IP"
LOGSTASH_PORT="5514"
PIHOLE_LOG="/var/log/pihole/pihole.log"

tail -F "$PIHOLE_LOG" | while read -r line; do
    echo "<190>$(date '+%b %e %H:%M:%S') $(hostname) pihole: $line" | \
    nc -w1 "$LOGSTASH_HOST" "$LOGSTASH_PORT"
done

That <190>... pihole: prefix wraps each line as a syslog message with the tag pihole. That tag is exactly what the pipeline routes on.

Script 2: blocked queries

Here's the wrinkle nobody warned me about: Pi-hole v6 doesn't write blocked queries to pihole.log at all. It answers them silently and records them only in its SQLite database. So the first script will never see a single blocked domain.

The fix is a second script that polls the database every 60 seconds for new blocked entries:

sudo nano /usr/local/bin/pihole-blocked-siem.sh

#!/bin/bash
# Ship Pi-hole blocked queries to Logstash

LOGSTASH_HOST="YOUR_ELASTIC_VM_IP"
LOGSTASH_PORT="5514"
DB="/etc/pihole/pihole-FTL.db"
STATE_FILE="/var/lib/pihole-siem-lastid"

# First run: start from the last 24 hours
if [ ! -f "$STATE_FILE" ]; then
    LAST_ID=$(sudo pihole-FTL sqlite3 "$DB" \
        "SELECT COALESCE(MIN(id),0) FROM queries WHERE timestamp > (strftime('%s','now') - 86400);")
    echo "$LAST_ID" > "$STATE_FILE"
fi

while true; do
    LAST_ID=$(cat "$STATE_FILE")

    RESULTS=$(sudo pihole-FTL sqlite3 "$DB" \
        "SELECT id, datetime(timestamp,'unixepoch','localtime'), domain, client, status
         FROM queries
         WHERE id > $LAST_ID
         AND status IN (2,7,9,12)
         ORDER BY id ASC;")

    if [ -n "$RESULTS" ]; then
        while IFS='|' read -r id timestamp domain client status; do
            MSG="<190>$timestamp $(hostname) pihole-blocked: id=$id domain=$domain client=$client status=$status"
            echo "$MSG" | nc -w1 "$LOGSTASH_HOST" "$LOGSTASH_PORT"
            echo "$id" > "$STATE_FILE"
        done <<< "$RESULTS"
    fi

    sleep 60
done

A few notes on this one:

• Status codes 2, 7, 9 and 12 are Pi-hole v6's blocked statuses (gravity, exact match, regex, and allow-list blocked).
• queries is a view in v6, not a table. It resolves domain and client names for you, so no JOINs are needed. Older guides that JOIN to domain_by_id fail with no such column.
• The <<< "$RESULTS" here-string matters. My first version piped the results into the while loop instead. That runs the loop in a subshell, so the state file updates were lost and the service processed one batch, then quietly exited.

Run both as services

Create a systemd unit for each so they start at boot and restart if they die. The DNS one:

sudo nano /etc/systemd/system/pihole-siem.service

[Unit]
Description=Pi-hole DNS log forwarder to Logstash
After=network.target pihole-FTL.service

[Service]
Type=simple
ExecStart=/usr/local/bin/pihole-siem.sh
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Copy it to pihole-blocked-siem.service, swapping the description and the ExecStart path. Then:

sudo chmod +x /usr/local/bin/pihole-siem.sh /usr/local/bin/pihole-blocked-siem.sh
sudo systemctl daemon-reload
sudo systemctl enable --now pihole-siem pihole-blocked-siem
sudo systemctl status pihole-siem pihole-blocked-siem

Generate some traffic to check it end to end:

dig google.com @YOUR_PIHOLE_IP

The $(hostname) in both scripts means a second Pi-hole needs no changes — copy the scripts across and its events land in the same index, told apart by hostname.

Source 2: Traefik

Traefik sees every web request to every service behind it, so it's where you spot failed requests, probing and broken backends. Its application log is valuable too — an expired Cloudflare API token was silently failing my certificate renewals for three days before I found it by accident. With the app log in the SIEM, that would have been obvious immediately.

As explained above, Traefik writes to local files and a separate script ships them. Traefik itself never depends on the SIEM.

Step 1: JSON logs to file

In your Traefik static config (traefik.yaml), add both blocks at the root indent level:

accessLog:
  format: json
  filePath: /var/log/traefik/access.log

log:
  level: INFO
  format: json
  filePath: /var/log/traefik/traefik.log

JSON is the key choice here. Logstash's json filter turns every field into a structured field automatically, so there are no grok patterns to write.

Two cautions before you save: it's accessLog with a capital L (lowercase accesslog crash-loops Traefik with field not found), and /var/log/traefik must be mounted to a folder on the host in your Traefik compose file. Static config isn't hot-reloaded, so restart Traefik:

sudo docker compose restart traefik
tail -3 /path/to/traefik/logs/access.log

You should see JSON lines appear.

Step 2: The shipping script

On the Docker host (install netcat-openbsd first if needed):

sudo nano /usr/local/bin/traefik-siem.sh

#!/bin/bash
# Forward Traefik access + application logs to Logstash

LOGSTASH_HOST="YOUR_ELASTIC_VM_IP"
LOGSTASH_PORT="5514"
ACCESS_LOG="/path/to/traefik/logs/access.log"
APP_LOG="/path/to/traefik/logs/traefik.log"

tail -F "$ACCESS_LOG" | while read -r line; do
    echo "<190>$(date '+%b %e %H:%M:%S') $(hostname) traefik: $line" | \
    nc -w1 "$LOGSTASH_HOST" "$LOGSTASH_PORT"
done &

tail -F "$APP_LOG" | while read -r line; do
    echo "<190>$(date '+%b %e %H:%M:%S') $(hostname) traefik-app: $line" | \
    nc -w1 "$LOGSTASH_HOST" "$LOGSTASH_PORT"
done &

wait

The two loops use different tags, traefik and traefik-app, so the pipeline can tell them apart. Both land in traefik-*, with app-log events tagged traefik_app. The trailing wait keeps the script alive for systemd.

Make it executable and wrap it in a systemd service exactly as for Pi-hole (traefik-siem.service, with After=network.target docker.service). If Logstash is down, the script just fails and systemd restarts it every five seconds until it's back. Traefik never notices.

Source 3: Authentik

Authentik is my SSO provider, so it answers "who logged in, and who tried to?" It already logs structured JSON to stdout, which makes it the easiest source to parse.

Before you apply this, know the trade-off: with the syslog driver, docker logs it stops working for these containers, because the output goes to Logstash instead of local disk. And if Logstash is unreachable when Authentik starts, Authentik can hang. I accepted that for Authentik; I wouldn't for anything critical.

In the Authentik compose file, add a logging block to both authentik-server and authentik-worker:

    logging:
      driver: syslog
      options:
        syslog-address: "tcp://YOUR_ELASTIC_VM_IP:5514"
        tag: "authentik"

The tag becomes [process][name] in Logstash, which is what routes these events to the Authentik branch. Logging driver changes only apply to new containers, so a plain restart isn't enough — recreate them:

sudo docker compose up -d --force-recreate

Confirm the driver is actually active:

sudo docker inspect authentik-server --format='{{json .HostConfig.LogConfig}}'

It should say syslog, not json-file. Then log in and out of Authentik once to generate some events.

One honest caveat: Authentik's operational logs are useful, but they're not a clean record of every login and failed login. For proper audit events, I ended up polling Authentik's API instead. That's a post of its own, and it's coming in Part 3, alongside the detection rules that depend on it.

Seeing it in Kibana

First, confirm the indices exist. On the Elastic VM:

curl -s -u elastic:YOUR_ELASTIC_PASSWORD http://localhost:9200/_cat/indices?v | grep -E "pihole|traefik|authentik"

You should see a dated index for each source, with a document count that grows every time you run it.

Now create a data view for each one: Stack Management → Data Views → Create data view. An index has to exist before you can create its view, so make sure each source has sent at least one event first.

Then open Discover, pick a data view and set the time range to the last 15 minutes. A few queries to start with:

One thing that confused me: Traefik's access log entries are always level info, even for a 404 or a 500. Use status codes to find failed requests, not the log level. Real errors live in the app log.

Click into any event, and you'll see the payoff of all that parsing: query_domain, client_ip, DownstreamStatus and friends as proper fields you can filter, count and chart.

Your first dashboard

Searching is great for answering a question you already have. A dashboard helps you spot the question in the first place. Three panels are enough to start.

Go to Dashboards → Create dashboard → Create visualisation. Each panel is built in Lens: pick a data view in the top left, pick a chart type, then set the axes in the right-hand panel.

Click Save and return after each panel, then save the dashboard itself. I called mine Security Overview.

Two things to know first:

• Use the .keyword version of text fields for Top values. The plain field is analysed for full-text search and can't be grouped.
• Expect surprises. My top-clients panel showed one machine making far more blocked lookups than anything else on the network. It turned out to be completely benign, but that's exactly the kind of thing you'd never notice without a dashboard.

Gotchas I hit (so you don't have to)

• Pi-hole v6 and rsyslog → the logs stop flowing after a few minutes. Tail pihole.log with a script instead.
• Blocked queries missing → v6 only records them in SQLite, never in pihole.log. Poll the database.
• Piping into a while loop → the loop runs in a subshell, state is lost and the service exits. Use a here-string.
• [process.name] in conditionals → matches nothing, silently. Use [process][name].
• Pipeline file named .config → Logstash ignores it. It must be .conf.
• A backup file in the pipeline folder → Logstash loads it too, both copies fight over port 5514 (Address already in use) and it won't start. Keep backups elsewhere.
• A mutate in the output section → Logstash crash-loops. Every parsing step belongs in filter.
• accesslog in lowercase → Traefik crash-loops. It's accessLog.
• Syslog driver on a restarted container → no change. Logging drivers apply only on --force-recreate.
• The syslog driver dependency is real. During a later incident, Logstash went down. When Authentik restarted it hung on its logging driver, and my wiki — which logs in through Authentik — returned 502 errors. Same root cause, three services deep. That's exactly why Traefik doesn't use it.
• Boot order → for the same reason, the Elastic VM now starts first after a host reboot, with a two-minute head start before the Docker host comes up.
• Firewall → if your sources are on different VLANs from the SIEM, you need rules allowing them to reach the Elastic VM on 5514. When nothing arrives, check docker compose logs logstash for new connection lines from each source's IP. No line means a network problem, not a pipeline problem.

Where we are

The SIEM finally has water in the pipes:

• Every DNS lookup and every blocked domain, from every Pi-hole
• Every web request through Traefik, plus its own errors
• Authentik's SSO activity

All of it parsed into real fields and searchable in Kibana.

Searching by hand is only the start, though. Next up: turning those searches into detection rules that alert you on their own — a brute-force login attempt, say, or a device suddenly talking to a domain nobody on the network has ever looked up before.

If you're building along, I'd love to hear which source fought you hardest. For me it was the blocked-queries script, by a distance.

Part 3 coming soon: detection rules and Authentik audit events.