Audit Logging: Tracking Data Changes for Compliance

Introduction

Every data change tells a story. Who modified a customer record at 3am? Which user accessed that sensitive configuration? Why did that pricing calculation change last month? Without audit logging, these questions become unanswerable—and regulators do not accept “we don’t know” as an answer.

Financial services, healthcare, payment processors—these industries require tracking who changed what, when, and often why. This guide covers the technical implementation of audit logging without the vendor fluff.

flowchart LR
    subgraph Capture["Audit Capture"]
        T[("Trigger<br/>Row Change")]
        CDC[("CDC<br/>Log Stream")]
        PGA[("pg_audit<br/>Statement")]
    end

    subgraph Pipeline["Processing Pipeline"]
        Queue[("Kafka /<br/>Log Stream")]
        Consumer[("Consumer<br/>Processor")]
        Agg[("Log<br/>Aggregator")]
    end

    subgraph Storage["Audit Storage"]
        AL[("Audit<br/>Log Table")]
        Partition[("Time<br/>Partition")]
        Archive[("S3 Glacier<br/>Archive")]
    end

    subgraph Analysis["SIEM / Analysis"]
        SIEM[("SIEM<br/>Platform")]
        Alert[("Real-time<br/>Alert")]
        Query[("Ad-hoc<br/>Query")]
    end

    T --> Queue
    CDC --> Queue
    PGA --> Queue
    Queue --> Consumer
    Consumer --> Agg
    Agg --> AL
    AL --> Partition
    Partition --> Archive
    Agg --> SIEM
    SIEM --> Alert
    SIEM --> Query

What to Log: The Four Ws

Effective audit logging captures the complete picture of data changes.

Who (Identity)

Log the user identity associated with each operation:

-- Include application user ID, not database session
-- Database sessions may be pooled (PostgreSQL pgBouncer, MySQL Proxy)
audit_log: {
    user_id: 12345,
    username: "john.smith@company.com",
    session_id: "sess_abc123",
    ip_address: "192.168.1.100",
    user_agent: "Mozilla/5.0..."
}

What (Operation)

Record the specific operation performed:

audit_log: {
    table_name: "customer_records",
    operation: "UPDATE",
    record_id: 789,
    old_values: { "credit_limit": 5000 },
    new_values: { "credit_limit": 10000 },
    query: "UPDATE customer_records SET credit_limit = 10000 WHERE id = 789"
}

When (Timestamp)

Use UTC timestamps with sufficient precision:

-- Microsecond precision for PostgreSQL
audit_log: {
    timestamp: "2026-03-26T14:30:00.123456Z",
    transaction_id: 1847293,
    sequence_in_transaction: 3
}

Additional Context

For a complete picture, consider logging:

• Application name: Which service initiated the change
• Request ID: Correlation ID for distributed tracing
• Reason code: Business justification when required
• Approval chain: Who approved privileged operations

Database Triggers for Audit Logging

Triggers provide the most direct way to capture row-level changes.

PostgreSQL Trigger Implementation

-- Create audit log table
CREATE TABLE audit_log (
    id BIGSERIAL PRIMARY KEY,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    table_name TEXT NOT NULL,
    operation TEXT NOT NULL,
    record_id TEXT,
    old_data JSONB,
    new_data JSONB,
    user_id INTEGER,
    username TEXT,
    application_name TEXT,
    transaction_id BIGINT,
    client_addr INET
);

-- Create the trigger function
CREATE OR REPLACE FUNCTION audit_trigger_function()
RETURNS TRIGGER AS $$
DECLARE
    audit_row audit_log;
    current_user_id INTEGER;
    current_username TEXT;
    current_app TEXT;
BEGIN
    -- Capture current user context
    SELECT usesysid::INTEGER, usename, COALESCE(current_setting('application_name', true), '')
    INTO current_user_id, current_username, current_app
    FROM pg_user WHERE usesysid = current_user;

    audit_row := ROW(
        nextval('audit_log_id_seq'),
        CURRENT_TIMESTAMP,
        TG_TABLE_NAME,
        TG_OP,
        CASE WHEN TG_OP = 'UPDATE' OR TG_OP = 'DELETE' THEN OLD.id::TEXT ELSE NEW.id::TEXT END,
        CASE WHEN TG_OP = 'UPDATE' OR TG_OP = 'DELETE' THEN to_jsonb(OLD) END,
        CASE WHEN TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN to_jsonb(NEW) END,
        current_user_id,
        current_username,
        current_app,
        txid_current(),
        inet_client_addr()
    );

    INSERT INTO audit_log VALUES (audit_row.*);
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- Attach trigger to a table
CREATE TRIGGER customers_audit
AFTER INSERT OR UPDATE OR DELETE ON customers
FOR EACH ROW EXECUTE FUNCTION audit_trigger_function();

Selective Column Auditing

Only audit sensitive columns to reduce storage and noise:

CREATE OR REPLACE FUNCTION audit_trigger_function()
RETURNS TRIGGER AS $$
DECLARE
    audit_row audit_log;
    sensitive_columns TEXT[] = ARRAY['ssn', 'credit_card', 'salary'];
    old_sensitive JSONB;
    new_sensitive JSONB;
BEGIN
    -- Only capture changes to sensitive columns
    IF TG_OP = 'UPDATE' THEN
        SELECT jsonb_object_agg(key, value) INTO old_sensitive
        FROM jsonb_each(to_jsonb(OLD))
        WHERE key = ANY(sensitive_columns);

        SELECT jsonb_object_agg(key, value) INTO new_sensitive
        FROM jsonb_each(to_jsonb(NEW))
        WHERE key = ANY(sensitive_columns);

        -- Skip if no sensitive columns changed
        IF old_sensitive IS DISTINCT FROM new_sensitive THEN
            INSERT INTO audit_log (
                created_at, table_name, operation, record_id,
                old_data, new_data, username, transaction_id
            ) VALUES (
                CURRENT_TIMESTAMP, TG_TABLE_NAME, TG_OP, OLD.id::TEXT,
                old_sensitive, new_sensitive, current_user, txid_current()
            );
        END IF;

        RETURN OLD;
    END IF;

    -- Full audit for INSERT and DELETE
    INSERT INTO audit_log (
        created_at, table_name, operation, record_id,
        old_data, new_data, username, transaction_id
    ) VALUES (
        CURRENT_TIMESTAMP, TG_TABLE_NAME, TG_OP,
        CASE WHEN TG_OP = 'DELETE' THEN OLD.id::TEXT ELSE NEW.id::TEXT END,
        to_jsonb(OLD), to_jsonb(NEW), current_user, txid_current()
    );

    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

MySQL Trigger Implementation

-- Create audit log table
CREATE TABLE audit_log (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    table_name VARCHAR(255) NOT NULL,
    operation ENUM('INSERT', 'UPDATE', 'DELETE') NOT NULL,
    record_id VARCHAR(255),
    old_data JSON,
    new_data JSON,
    user_id INT,
    username VARCHAR(255),
    client_ip VARCHAR(45)
);

-- Create the trigger
DELIMITER //

CREATE TRIGGER customers_audit_insert
AFTER INSERT ON customers
FOR EACH ROW
BEGIN
    INSERT INTO audit_log (table_name, operation, record_id, new_data, user_id, username, client_ip)
    VALUES (
        'customers',
        'INSERT',
        NEW.id,
        JSON_OBJECT('name', NEW.name, 'email', NEW.email),
        @current_user_id,
        @current_username,
        @current_client_ip
    );
END//

CREATE TRIGGER customers_audit_update
AFTER UPDATE ON customers
FOR EACH ROW
BEGIN
    INSERT INTO audit_log (table_name, operation, record_id, old_data, new_data, user_id, username, client_ip)
    VALUES (
        'customers',
        'UPDATE',
        OLD.id,
        JSON_OBJECT('name', OLD.name, 'email', OLD.email),
        JSON_OBJECT('name', NEW.name, 'email', NEW.email),
        @current_user_id,
        @current_username,
        @current_client_ip
    );
END//

CREATE TRIGGER customers_audit_delete
AFTER DELETE ON customers
FOR EACH ROW
BEGIN
    INSERT INTO audit_log (table_name, operation, record_id, old_data, user_id, username, client_ip)
    VALUES (
        'customers',
        'DELETE',
        OLD.id,
        JSON_OBJECT('name', OLD.name, 'email', OLD.email),
        @current_user_id,
        @current_username,
        @current_client_ip
    );
END//

DELIMITER ;

Change Data Capture (CDC)

For high-volume systems, triggers can become a bottleneck. CDC offers an alternative approach using the database’s transaction log.

PostgreSQL CDC with Debezium

# Docker Compose for Debezium PostgreSQL connector
version: "3"
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: mydb
    command:
      - "postgres"
      - "-c"
      - "wal_level=logical"
      - "-c"
      - "max_wal_senders=10"
      - "-c"
      - "max_replication_slots=10"

  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on: [zookeeper]
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

  debezium:
    image: debezium/connect:2.3
    depends_on: [kafka, postgres]
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: debezium_configs
      OFFSET_STORAGE_TOPIC: debezium_offsets
      STATUS_STORAGE_TOPIC: debezium_statuses

// Debezium connector configuration
{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "mydb",
    "topic.prefix": "dbserver1",
    "table.include.list": "public.customers,public.orders",
    "plugin.name": "pgoutput",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

CDC Advantages over Triggers

CDC and triggers take different approaches to the same problem, and the architectural difference matters more as write volume increases. The trade-offs below should help you pick the right one for your situation.

CDC captures changes by reading the database’s transaction log rather than executing trigger code on each row modification. When an UPDATE statement modifies 10,000 rows, a trigger-based approach fires the audit function 10,000 times synchronously. CDC instead miners the write-ahead log asynchronously, producing change events without blocking the write path.

The near real-time characteristic stems from log-based streaming. Transaction log changes are written to disk almost immediately after a transaction commits, and CDC connectors tail this log continuously. The lag between a database write and an audit event appearing in the stream is typically measured in milliseconds, compared to trigger-based audit which is limited to synchronous execution within the transaction.

CDC’s multi-system propagation comes naturally from the streaming architecture. A single change event published to Kafka can be consumed by multiple independent consumers: one for the audit log table, one for a search index update, one for a cache invalidation, one for a data warehouse sync. This fan-out would require separate trigger logic for each destination with trigger-based auditing, but CDC handles it through topic subscriptions.

Schema evolution handling works because CDC tools track schema changes separately and emit schema change events alongside data change events. Debezium, for example, records the schema before and after each DDL operation, allowing consumers to interpret change events from older schema versions correctly. This is significantly more robust than trigger-based approaches which require manual trigger updates whenever table schemas change.

CDC Limitations

The setup complexity is substantial and operational. A trigger-based audit system requires SQL knowledge and per-table trigger creation. A CDC-based audit system requires Kafka expertise (or a cloud streaming service), connector configuration, consumer development, and dead letter queue handling. Teams without streaming experience should budget 2-4 weeks for initial setup and testing versus 1-2 days for triggers. The operational burden continues after deployment: Kafka cluster maintenance, connector monitoring, consumer group lag alerting, and schema registry management.

Kafka consumer lag during outages is the primary availability risk. If the Kafka broker becomes unavailable or the consumer falls behind due to a processing error, audit events accumulate in the topic. During the outage window, audit events are not being processed into the audit log table. For compliance scenarios requiring synchronous audit guarantees, this eventual consistency model is problematic. Mitigation: monitor consumer lag, alert aggressively, and design consumers for idempotent reprocessing so lag catchup does not produce duplicate audit entries.

The initial snapshot requirement exists because CDC connectors can only capture changes from the moment they start. Any data already in the database before the connector begins running is invisible to CDC. Populating the audit trail for existing data requires a one-time snapshot operation: the connector reads all existing rows and emits them as INSERT events, then switches to log-based capture for new changes. For large tables (hundreds of millions of rows), this snapshot can take days and generate significant Kafka traffic.

PostgreSQL pg_audit

PostgreSQL’s pg_audit extension provides standardized, comprehensive audit logging:

-- Enable pg_audit (requires PostgreSQL installation with extension)
CREATE EXTENSION pgaudit;

-- Configure pg_audit in postgresql.conf
-- pgaudit.log = 'read, write, function, role, ddl'
-- pgaudit.log_catalog = on
-- pgaudit.log_relation = on
-- pgaudit.log_statement_once = off

-- Example audit log output (in PostgreSQL log)
-- 2026-03-26 14:30:00.123 UTC [12345] user@db LOG:  AUDIT: SESSION,1,1,READ,SELECT,,,SELECT * FROM customers WHERE id = 1; <pid=12345>
-- 2026-03-26 14:30:01.456 UTC [12345] user@db LOG:  AUDIT: SESSION,1,2,WRITE,INSERT,customer_records,"INSERT INTO customer_records ...",<not logged>

pg_audit vs. Custom Triggers

When to Use / When Not to Use Each Audit Approach

Use Triggers when:

• You need row-level capture of old/new values
• Per-table auditing is acceptable overhead
• You need deterministic capture within transaction context

Do not use Triggers when:

• Your system writes millions of rows per second
• You need cross-table correlation in a single audit event
• You want minimal performance impact on write throughput

Use CDC when:

• You need audit data feeding multiple downstream systems
• Write throughput cannot tolerate trigger overhead
• You need near-real-time audit event streaming

Do not use CDC when:

• Your infrastructure budget doesn’t cover Kafka/connector
• You need synchronous audit within the transaction that made the change

Use pg_audit when:

• You need standardized DDL and session auditing on PostgreSQL
• Compliance requires evidence of all statement executions

Do not use pg_audit alone when:

• You need row-level old/new value capture (pg_audit doesn’t provide this)

Audit Approach Trade-offs

Production Failure Scenarios

Capacity Estimation: Log Volume Per Transaction and Storage Planning

Audit logging generates substantial data volume. Planning storage requires understanding your transaction rate and the size of each logged change.

Log volume formula:

audit_log_bytes_per_transaction = avg_changed_columns × avg_column_size × (old_value_json_overhead + new_value_json_overhead)
daily_audit_log_volume = transactions_per_day × changes_per_transaction × audit_log_bytes_per_change

For an OLTP system with 50,000 daily transactions, each modifying 5 columns averaging 50 bytes each (as JSON):

• Old + new value as JSON: ~2× overhead for JSON wrapper and field names
• Bytes per change: 5 × 50 × 2.5 = 625 bytes per modified row
• Audit log per transaction: 3 modified rows × 625 = ~1.9KB
• Daily log volume: 50,000 × 1.9KB = ~95MB/day
• Annual: 95MB × 365 = ~35GB/year

This assumes the audit log stores both old and new values as JSON. Storing old values only (for INSERT/DELETE) or new values only (for UPDATE) halves the volume. Storing diffs instead of full values reduces it further.

CDC-based audit log volume (for Kafka-based CDC):

kafka_message_bytes = avg_row_width × number_of_columns × 2 (old+new) × json_overhead × schema_compression
daily_kafka_volume = cdc_events_per_day × avg_event_size

For the same 50K transactions/day with CDC capturing all columns (50 columns, 500 bytes each): 50K × 50 × 500 × 2.5 = ~3.1GB/day — significantly larger than trigger-based audit logging because CDC captures entire rows, not just changed columns.

Observability Hooks: CDC Lag Metrics and Audit Log Completeness Checks

Key audit logging metrics: CDC lag, log completeness, and write latency.

-- PostgreSQL: check if any audit log entries are missing (gap detection)
SELECT
    last_log_id,
    last_log_time,
    EXTRACT(EPOCH FROM (now() - last_log_time)) AS gap_seconds
FROM audit_log_sequence
WHERE last_log_time < now() - INTERVAL '5 minutes';

-- Verify audit log completeness for a critical table
SELECT
    audit_table_name,
    COUNT(*) AS audit_entries,
    COUNT(DISTINCT DATE(occurred_at)) AS days_covered,
    MIN(occurred_at) AS oldest_entry,
    MAX(occurred_at) AS newest_entry
FROM audit_log
WHERE audit_table_name = 'financial_transactions'
GROUP BY audit_table_name;

# Critical: CDC lag exceeds threshold (data freshness at risk)
- alert: CdcLagExceedsThreshold
  expr: cdc_consumer_lag_messages > 10000
  for: 10m
  labels:
    severity: critical
  annotations:
    summary: "CDC consumer lag {{ $value }} messages on {{ $labels.topic }}"

# Critical: Audit log write latency spike (could indicate log writer bottleneck)
- alert: AuditLogWriteLatencyHigh
  expr: histogram_quantile(0.95, audit_log_write_duration_seconds) > 1.0
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "Audit log P95 write latency {{ $value }}s exceeds 1 second"

# Warning: Audit log entry count anomaly (could indicate missed entries)
- alert: AuditLogVolumeAnomaly
  expr: rate(audit_log_entries_total[1h]) < 0.5 * avg_over_time(rate(audit_log_entries_total[1h])[24h:1h])
  for: 1h
  labels:
    severity: warning
  annotations:
    summary: "Audit log volume {{ $value }} is 50% below average"

# Critical: Audit log storage approaching capacity
- alert: AuditLogStorageCritical
  expr: (audit_log_storage_used_bytes / audit_log_storage_quota_bytes) > 0.85
  for: 15m
  labels:
    severity: critical
  annotations:
    summary: "Audit log storage at {{ $value | humanizePercentage }} capacity"

Quick Recap Checklist

Use this checklist when designing or reviewing audit logging for a database system:

• Audit log captures who, what, when, and context for every data change
• Application user ID (not just database session) is captured in audit entries
• Row-level old/new values are captured for INSERT/UPDATE/DELETE operations
• Audit table is partitioned by time for efficient retention management
• Retention policy satisfies compliance requirements (PCI-DSS, HIPAA, GDPR)
• Completeness monitoring is in place with gap detection alerts
• CDC or trigger overhead has been benchmarked under realistic write load
• Selective column auditing configured for high-volume tables
• Audit log queries use appropriate indexes for performance
• Archival strategy defined for moving old partitions to cold storage

Log Aggregation and Analysis

Raw audit logs aren’t audit logs until they’re analyzed. Collecting logs without querying them is just storage expense.

Centralized Log Collection

import logging
import json
from datetime import datetime
import psycopg2
from kafka import KafkaProducer

class AuditLogger:
    def __init__(self, kafka_servers: list, audit_topic: str):
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        self.audit_topic = audit_topic

    def log_change(
        self,
        table: str,
        operation: str,
        record_id: str,
        old_values: dict,
        new_values: dict,
        context: dict
    ):
        event = {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'table': table,
            'operation': operation,
            'record_id': record_id,
            'old_values': old_values,
            'new_values': new_values,
            **context
        }

        self.producer.send(self.audit_topic, event)

    def close(self):
        self.producer.flush()
        self.producer.close()

Query Patterns for Audit Analysis

-- Who modified this record and when?
SELECT
    created_at,
    username,
    operation,
    old_data,
    new_data
FROM audit_log
WHERE table_name = 'customers'
  AND record_id = '123'
  AND (old_data->>'email') IS DISTINCT FROM (new_data->>'email')
ORDER BY created_at DESC;

-- What changed in the last 24 hours?
SELECT
    table_name,
    operation,
    COUNT(*) as change_count,
    COUNT(DISTINCT username) as users_involved
FROM audit_log
WHERE created_at > CURRENT_TIMESTAMP - INTERVAL '24 hours'
GROUP BY table_name, operation
ORDER BY change_count DESC;

-- Flag unusual access patterns
SELECT
    username,
    COUNT(*) as access_count,
    COUNT(DISTINCT table_name) as tables_accessed,
    MAX(created_at) as last_access
FROM audit_log
WHERE created_at > CURRENT_TIMESTAMP - INTERVAL '1 hour'
GROUP BY username
HAVING COUNT(*) > 1000;  -- Flag unusual volume

Retention Policies

Audit logs have regulatory lifecycles:

• PCI-DSS: 1 year minimum, 3 years recommended
• HIPAA: 6 years from creation or last effective date
• GDPR: As long as necessary for processing purposes
• SOC 2: 1 year minimum

Archival Strategy

-- Partition audit log by month for efficient retention management
CREATE TABLE audit_log (
    id BIGSERIAL,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    table_name TEXT NOT NULL,
    operation TEXT NOT NULL,
    record_id TEXT,
    old_data JSONB,
    new_data JSONB,
    user_id INTEGER,
    username TEXT,
    PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);

-- Create monthly partitions
CREATE TABLE audit_log_2026_03 PARTITION OF audit_log
    FOR VALUES FROM ('2026-03-01') TO ('2026-04-01');

CREATE TABLE audit_log_2026_04 PARTITION OF audit_log
    FOR VALUES FROM ('2026-04-01') TO ('2026-05-01');

-- Retention: drop partitions older than 2 years
-- This is much faster than DELETE
DROP TABLE audit_log_2024_03;

import boto3
from datetime import datetime, timedelta

def archive_old_partitions(db_config: dict, retention_days: int):
    """Archive audit log partitions to S3 before dropping"""
    s3 = boto3.client('s3')

    cutoff_date = datetime.now() - timedelta(days=retention_days)

    # Export partitions older than retention
    # In production, use proper export mechanism
    for partition in get_partitions_before(db_config, cutoff_date):
        # Export to Parquet for efficient storage
        export_to_parquet(partition)

        # Upload to S3 Glacier for long-term storage
        s3.upload_file(
            f'/tmp/{partition}.parquet',
            'audit-log-archive',
            f'archive/{partition}.parquet'
        )

        # Drop the partition from database
        drop_partition(partition)

Security and Compliance Notes

Audit records can expose more than ordinary application logs. Store only the fields needed to explain a change; redact or tokenize secrets and personal data instead of copying full row snapshots. Give read and export access to named roles, and keep the audit writer from changing or deleting existing entries. Encrypt logs in transit and at rest, with access to encryption keys managed separately.

Use append-only storage or an immutable archive when the audit requirement calls for tamper resistance. Hash-chain entries or sign batches, verify them on a schedule, and alert on broken chains, missing entries, sudden drops in event volume, or disabled capture. Apply the retention and deletion schedule to archives as well as live tables, and document any legal holds.

Interview Questions

Further Reading

• Constraint Enforcement — Data integrity fundamentals
• Encryption at Rest — Protecting stored audit data

Conclusion

Audit logging is infrastructure, not afterthought. Capture the complete picture—who, what, when, and context—at the database level where it cannot be bypassed. Use triggers for straightforward per-table tracking, consider CDC for high-volume systems, and leverage pg_audit for standardized PostgreSQL logging.

Aggregate logs centrally for analysis, and implement retention policies that satisfy your regulatory requirements without accumulating unnecessary data.

For related reading, explore our constraint enforcement guide for maintaining data integrity, or dive into logging best practices for broader operational logging patterns.