Audit Logging: Tracking Data Changes for Compliance

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

Introduction

Audit logging captures every data change so you can answer “who changed what, when, and why.” This is not just a compliance checkbox — it is how you diagnose production incidents, prove regulatory compliance, and recover from accidental data deletion. Without audit logging, data changes are invisible unless you have row-level history or external logging.

This guide covers what to log (the four Ws: who, what, when, and where), capture mechanisms (triggers, change data capture, pg_audit), log storage and retention, and how to implement audit logging that does not become a performance bottleneck on your write path.

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_text(to_jsonb(OLD))
        WHERE key = ANY(sensitive_columns);

        SELECT jsonb_object_agg(key, value) INTO new_sensitive
        FROM jsonb_each_text(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 VALUES (
                nextval('audit_log_id_seq'),
                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 VALUES (
        nextval('audit_log_id_seq'),
        CURRENT_TIMESTAMP,
        TG_TABLE_NAME,
        TG_OP,
        COALESCE(NEW.id::TEXT, OLD.id::TEXT),
        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

• Lower overhead: No per-table trigger execution
• Near real-time: Changes stream as they happen, not after commit
• Multi-system propagation: Feed to multiple sinks (data warehouse, search index, cache)
• Schema evolution: CDC handles schema changes more gracefully

CDC Limitations

• Setup complexity: Requires Kafka, connector configuration, consumer infrastructure
• Latency during failures: Kafka consumer lag during outages
• Initial snapshot: Full table copy required to bootstrap

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
-- pgaudit.role = audit_user

-- Create dedicated audit role
CREATE ROLE audit_user;
GRANT EXECUTE ON FUNCTION pg_logonal_session TO audit_user;

-- 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

Feature	pg_audit	Custom Triggers
Setup complexity	Low	High
Log format	Standardized	Custom
Performance	Optimized	Variable
Flexibility	Limited to configured statements	Full control
Cross-table operations	Partial	Full
DDL auditing	Yes	Limited

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

Dimension	Database Triggers	CDC (Debezium)	pg_audit	Application-Level
Row-level old/new values	Yes	Yes	No	Yes
DDL auditing	No	No	Yes	Limited
Performance impact	Moderate to high	Low	Low	Low
Setup complexity	Medium	High	Low	Medium
Real-time streaming	No	Yes	No	Yes
Schema evolution handling	Manual trigger update	Automatic	Automatic	Manual
Infrastructure needed	None extra	Kafka cluster	Extension only	Logging framework
Best for	Per-table compliance	High-volume audit	Standard DDL/Session	Business-level audit

Production Failure Scenarios

Failure	Impact	Mitigation
Trigger causing deadlocks on high-write table	Write throughput collapses	Move auditing to application layer, benchmark before deploying
CDC consumer lag growing unbounded	Audit events delayed, compliance gap	Monitor lag metric, alert at threshold, scale consumers
Audit log table filling disk	Database stops writing	Partition audit table by time, enforce retention automatically
pg_audit flooding log storage	Storage costs spike	Set appropriate log levels, rotate aggressively
Selective column audit missing updates	Incomplete audit trail	Test with updates that skip audited columns

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)

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.