Project 4 (Capstone): Log Forensics & Incident Report Generator

Every previous project produced or consumed logs. This one closes the loop: it takes the raw output of a bad hour and produces the document you would paste into a postmortem — and it exercises nearly every technique in the module.

The Requirements

#	Requirement	Skills used
1	Accept a time window and filter every source to it	getopts, `date`, awk
2	Ingest ≥2 formats into one normalised stream	awk, sed, process substitution
3	Merged chronological timeline across sources	sort, associative arrays
4	Error rate per minute vs a computed baseline	awk aggregation, arithmetic
5	Top endpoints, clients and status codes	awk, sort, uniq
6	First-occurrence detection	`grep -m1`, awk
7	Markdown report to stdout, diagnostics to stderr	redirection discipline
8	Graceful handling of a missing or unreadable source	error handling
9	Testable functions	`main` guard, bats

Step 1: Interface

#!/usr/bin/env bash
#
# forensics.sh — correlate logs across sources into an incident report.
#
set -Eeuo pipefail

readonly SCRIPT_NAME=${0##*/}
readonly VERSION="1.0.0"
readonly E_USAGE=2 E_INPUT=3

START_TIME=""
END_TIME=""
ACCESS_LOG="${ACCESS_LOG:-/var/log/nginx/access.log}"
APP_LOG="${APP_LOG:-/var/log/app/application.log}"
UNITS="${UNITS:-nginx,postgresql}"
BASELINE_MINUTES="${BASELINE_MINUTES:-60}"
OUTPUT=""

info() { printf '%s [INFO ] %s\n' "$(date -Iseconds)" "$*" >&2; }
warn() { printf '%s [WARN ] %s\n' "$(date -Iseconds)" "$*" >&2; }
die()  { printf '%s [ERROR] %s\n' "$(date -Iseconds)" "$*" >&2; exit "${2:-1}"; }

usage() {
    cat <<EOF
${SCRIPT_NAME} ${VERSION} — build an incident report from logs

USAGE
    ${SCRIPT_NAME} -s START -e END [-o FILE]

OPTIONS
    -s START   window start, any format 'date -d' accepts  (e.g. '2026-08-05 14:00')
    -e END     window end
    -o FILE    write the report to FILE (default: stdout)
    -a PATH    nginx access log      (default: ${ACCESS_LOG})
    -l PATH    application log       (default: ${APP_LOG})
    -u LIST    comma-separated systemd units (default: ${UNITS})
    -h         this help

EXAMPLE
    ${SCRIPT_NAME} -s '2026-08-05 14:00' -e '2026-08-05 15:00' -o incident.md
EOF
}

All diagnostics go to stderr so the report itself can be redirected cleanly — forensics.sh -s ... -e ... > report.md must produce a clean file even while progress messages appear on the terminal.

Step 2: Normalising Time

Every source uses a different timestamp format. Convert everything to epoch seconds once, at the boundary.

to_epoch() {
    date -d "$1" +%s 2>/dev/null || die "unparseable time: $1" "$E_USAGE"
}

# nginx: 05/Aug/2026:14:23:11 +0000
nginx_ts_to_epoch() {
    local ts=${1#[}                       # strip leading [
    ts=${ts%%\ *}                         # drop the timezone
    date -d "$(echo "$ts" | sed 's|\([0-9]\{2\}\)/\([A-Za-z]\{3\}\)/\([0-9]\{4\}\):|\3-\2-\1 |')" +%s 2>/dev/null || echo 0
}

Doing this per line in bash would fork twice per record. The trick is to push the conversion into awk, which has mktime:

readonly AWK_NGINX_EPOCH='
function ts_epoch(field,   d, m, months, parts) {
    months = "JanFebMarAprMayJunJulAugSepOctNovDec"
    gsub(/[\[\]]/, "", field)
    split(field, parts, /[\/:]/)
    m = int((index(months, parts[2]) + 2) / 3)
    return mktime(parts[3] " " m " " parts[1] " " parts[4] " " parts[5] " " parts[6])
}'

One awk process converts a million lines. That is the difference between a report in two seconds and a report in twenty minutes.

Step 3: Ingest and Normalise

Every source is reduced to the same tab-separated shape: epoch <TAB> source <TAB> severity <TAB> message.

ingest_nginx() {
    local start=$1 end=$2
    [[ -r $ACCESS_LOG ]] || { warn "unreadable, skipping: ${ACCESS_LOG}"; return 0; }

    awk -v start="$start" -v end="$end" "$AWK_NGINX_EPOCH"'
    {
        e = ts_epoch($4)
        if (e < start || e > end) next
        sev = ($9 >= 500) ? "ERROR" : ($9 >= 400) ? "WARN" : "INFO"
        printf "%d\tnginx\t%s\t%s %s %s %s\n", e, sev, $1, $6, $7, $9
    }' "$ACCESS_LOG"
}

ingest_journal() {
    local start=$1 end=$2
    local unit
    for unit in ${UNITS//,/ }; do
        journalctl -u "$unit" \
            --since "@${start}" --until "@${end}" \
            -o short-unix --no-pager 2>/dev/null |
        awk -v u="$unit" '
        {
            split($1, t, ".")
            sev = /error|fail|fatal|panic/ ? "ERROR" : /warn/ ? "WARN" : "INFO"
            msg = ""
            for (i = 4; i <= NF; i++) msg = msg $i " "
            printf "%d\t%s\t%s\t%s\n", t[1], u, sev, msg
        }' || warn "journalctl failed for unit ${unit}"
    done
}

ingest_app() {
    local start=$1 end=$2
    [[ -r $APP_LOG ]] || { warn "unreadable, skipping: ${APP_LOG}"; return 0; }

    awk -v start="$start" -v end="$end" '
    {
        gsub(/[-:T]/, " ", $1)
        e = mktime($1 " " $2)
        if (e < start || e > end) next
        printf "%d\tapp\t%s\t%s\n", e, $3, substr($0, index($0, $4))
    }' "$APP_LOG" 2>/dev/null || warn "could not parse ${APP_LOG}"
}

Each ingester degrades gracefully: a missing log warns and returns 0 rather than aborting. An incident report built from two of three sources is far more useful than no report because one file was rotated away.

Step 4: The Merged Timeline

build_timeline() {
    local start=$1 end=$2
    {
        ingest_nginx   "$start" "$end"
        ingest_journal "$start" "$end"
        ingest_app     "$start" "$end"
    } | sort -n -k1,1
}

Grouping the three with { } and piping once means a single sort merges all sources chronologically. This is the correlation step, and it is four lines — because every ingester agreed on one output shape.

Step 5: Analysis

error_rate_per_minute() {
    local timeline=$1
    awk -F'\t' '
        { minute = strftime("%H:%M", $1); total[minute]++ }
        $3 == "ERROR" { errors[minute]++ }
        END {
            for (m in total)
                printf "%s\t%d\t%d\t%.1f\n", m, total[m], errors[m]+0, (errors[m]+0)*100/total[m]
        }' "$timeline" | sort
}

detect_anomaly() {
    local timeline=$1 baseline_start=$2 window_start=$3
    awk -F'\t' -v bstart="$baseline_start" -v wstart="$window_start" '
        $1 >= bstart && $1 < wstart { btotal++; if ($3 == "ERROR") berr++ }
        $1 >= wstart                { wtotal++; if ($3 == "ERROR") werr++ }
        END {
            brate = (btotal > 0) ? berr * 100 / btotal : 0
            wrate = (wtotal > 0) ? werr * 100 / wtotal : 0
            factor = (brate > 0) ? wrate / brate : (wrate > 0 ? 999 : 0)
            printf "%.2f\t%.2f\t%.1f\t%d\t%d\n", brate, wrate, factor, berr+0, werr+0
        }' "$timeline"
}

first_occurrence() {
    local timeline=$1 pattern=$2
    awk -F'\t' -v p="$pattern" '$4 ~ p { print strftime("%Y-%m-%d %H:%M:%S", $1) "\t" $2 "\t" $4; exit }' "$timeline"
}

top_n() {
    local timeline=$1 field=$2 n=${3:-10}
    awk -F'\t' -v f="$field" '$3 == "ERROR" { split($4, a, " "); print a[f] }' "$timeline" |
        sort | uniq -c | sort -rn | head -"$n"
}

detect_anomaly is the piece that turns data into a finding. “412 errors” is not a conclusion; “8.4× the baseline error rate” is.

Step 6: The Report

generate_report() {
    local timeline=$1 start=$2 end=$3 baseline_start=$4

    local brate wrate factor berr werr
    IFS=$'\t' read -r brate wrate factor berr werr < <(detect_anomaly "$timeline" "$baseline_start" "$start")

    local total_events
    total_events=$(wc -l < "$timeline")

    cat <<EOF
# Incident Report

**Generated:** $(date -Iseconds)
**Host:** $(hostname -f 2>/dev/null || hostname)
**Window:** $(date -d "@${start}" '+%Y-%m-%d %H:%M:%S') → $(date -d "@${end}" '+%Y-%m-%d %H:%M:%S')
**Events analysed:** ${total_events}

## Summary

| Metric | Baseline (${BASELINE_MINUTES}m before) | Incident window |
| :--- | ---: | ---: |
| Error rate | ${brate}% | **${wrate}%** |
| Error count | ${berr} | **${werr}** |
| Deviation | — | **${factor}×** |

$(awk -v f="$factor" 'BEGIN {
    if (f >= 5)      print "> **Significant anomaly.** The error rate is " f "x the baseline."
    else if (f >= 2) print "> **Elevated errors** at " f "x the baseline."
    else             print "> No significant deviation from baseline detected."
}')

## First Occurrence

\`\`\`
$(first_occurrence "$timeline" "5[0-9][0-9]|ERROR|error" || echo "none found")
\`\`\`

## Error Rate by Minute

| Time | Events | Errors | Rate |
| :--- | ---: | ---: | ---: |
$(error_rate_per_minute "$timeline" | awk -F'\t' '$3 > 0 {printf "| %s | %d | %d | %.1f%% |\n", $1, $2, $3, $4}')

## Top Offending Endpoints

\`\`\`
$(top_n "$timeline" 3 10)
\`\`\`

## Top Clients by Error Count

\`\`\`
$(top_n "$timeline" 1 10)
\`\`\`

## Events by Source

\`\`\`
$(awk -F'\t' '{print $2}' "$timeline" | sort | uniq -c | sort -rn)
\`\`\`

## Timeline (errors only, first 50)

\`\`\`
$(awk -F'\t' '$3 == "ERROR" {print strftime("%H:%M:%S", $1), "[" $2 "]", $4}' "$timeline" | head -50)
\`\`\`

## System Context

\`\`\`
$(uptime)
$(free -h | head -2)
$(df -hP | awk 'NR==1 || int($5) >= 80')
\`\`\`

---
*Generated by ${SCRIPT_NAME} ${VERSION}*
EOF
}

Every section answers a postmortem question: what happened, how bad against normal, when did it start, what was involved, what else was going on.

Step 7: Main

main() {
    while getopts ":s:e:o:a:l:u:h" opt; do
        case $opt in
            s) START_TIME=$OPTARG ;;
            e) END_TIME=$OPTARG ;;
            o) OUTPUT=$OPTARG ;;
            a) ACCESS_LOG=$OPTARG ;;
            l) APP_LOG=$OPTARG ;;
            u) UNITS=$OPTARG ;;
            h) usage; exit 0 ;;
            \?) die "unknown option -${OPTARG}" "$E_USAGE" ;;
            :)  die "-${OPTARG} requires an argument" "$E_USAGE" ;;
        esac
    done

    [[ -n $START_TIME && -n $END_TIME ]] || { usage >&2; exit "$E_USAGE"; }

    local start end baseline_start
    start=$(to_epoch "$START_TIME")
    end=$(to_epoch "$END_TIME")
    (( end > start )) || die "end must be after start" "$E_USAGE"
    baseline_start=$(( start - BASELINE_MINUTES * 60 ))

    local timeline
    timeline=$(mktemp) || die "cannot create temp file" "$E_INPUT"
    trap 'rm -f "$timeline"' EXIT

    info "collecting events from ${BASELINE_MINUTES}m baseline + incident window"
    build_timeline "$baseline_start" "$end" > "$timeline"

    local n; n=$(wc -l < "$timeline")
    (( n > 0 )) || die "no events found in the window — check paths and times" "$E_INPUT"
    info "collected ${n} events"

    if [[ -n $OUTPUT ]]; then
        generate_report "$timeline" "$start" "$end" "$baseline_start" > "$OUTPUT"
        info "report written to ${OUTPUT}"
    else
        generate_report "$timeline" "$start" "$end" "$baseline_start"
    fi
}

if [[ "${BASH_SOURCE[0]}" == "${0}" ]]; then
    main "$@"
fi

Step 8: Tests

The main guard means the analysis functions are sourceable and testable:

# tests/test_forensics.bats
setup() {
    load '../forensics.sh'
    export FIXTURE="${BATS_TEST_TMPDIR}/timeline"
    printf '%s\n' \
        "1000\tnginx\tINFO\t10.0.0.1 GET /a 200" \
        "1060\tnginx\tERROR\t10.0.0.2 GET /b 500" \
        "1120\tnginx\tERROR\t10.0.0.2 GET /b 500" > "$FIXTURE"
}

@test "detect_anomaly computes the incident error rate" {
    run detect_anomaly "$FIXTURE" 900 1050
    [ "$status" -eq 0 ]
    [[ "$output" == *"100.00"* ]]      # both post-1050 events are errors
}

@test "first_occurrence returns the earliest match only" {
    run first_occurrence "$FIXTURE" "500"
    [ "$(wc -l <<< "$output")" -eq 1 ]
}

@test "to_epoch rejects garbage" {
    run to_epoch "not a date"
    [ "$status" -ne 0 ]
}

Verifying Your Work

# Against a real window on any host with nginx
./forensics.sh -s '1 hour ago' -e 'now' -o /tmp/report.md
less /tmp/report.md

# Missing sources must degrade, not crash
./forensics.sh -a /nonexistent -s '1 hour ago' -e 'now' | head -30

# Redirection discipline: the report file must be clean
./forensics.sh -s '1 hour ago' -e 'now' > /tmp/clean.md 2>/tmp/diagnostics.log
head -5 /tmp/clean.md          # starts with "# Incident Report"

bats tests/

Extensions worth building: flag the deploy that preceded the window by reading the audit log from Project 3; correlate with the monitor’s state directory from Project 2 to show which alerts fired; add --format json for ingestion into a dashboard; add a --compare-to flag that diffs two windows.

The lesson to take away: the value is not in any single command — it is in the normalisation. Three sources with different formats became one analysable stream because every ingester agreed on epoch → source → severity → message. That decision is what made correlation a sort and analysis a handful of awk programs, and it is the same decision that makes any log platform work.