Database Transaction Monitor

Database Transaction Monitor

Monitor database transaction performance, detect long-running transactions, identify lock contention, track rollback rates, and automatically alert on transaction anomalies for production database health.

When to Use This Command

Use /txn-monitor when you need to:

• Detect and kill long-running transactions blocking other queries
• Monitor lock wait times and identify deadlock patterns
• Track transaction rollback rates for error analysis
• Alert on isolation level anomalies (phantom reads, dirty reads)
• Analyze transaction throughput and latency trends
• Investigate application connection leak issues

DON'T use this when:

• Database has minimal transaction load (<100 TPS)
• All transactions complete within milliseconds
• Looking for query optimization (use query optimizer instead)
• Investigating data corruption (use audit logger instead)

Design Decisions

This command implements real-time transaction monitoring with automated alerting because:

• Long-running transactions (>30s) block other queries and cause performance degradation
• Lock contention detection prevents cascade failures
• Rollback rate monitoring identifies application bugs early
• Automatic alerts reduce MTTR (Mean Time To Resolution)
• Historical trend analysis enables capacity planning

Alternative considered: Periodic manual checks

• No automated alerting on issues
• Relies on humans checking dashboards
• Slower incident response
• Recommended only for development environments

Alternative considered: Database log parsing

• Post-mortem analysis only
• No real-time alerts
• Requires custom log parsing logic
• Recommended for compliance/audit purposes

Prerequisites

Before running this command:

1. Database monitoring permissions (pg_monitor role or PROCESS privilege)
2. Access to pg_stat_activity (PostgreSQL) or performance_schema (MySQL)
3. Alerting infrastructure (Slack, PagerDuty, email)
4. Monitoring data retention strategy (metrics database or time-series DB)
5. Runbook for common transaction issues

Implementation Process

Step 1: Enable Transaction Monitoring

Configure database to track transaction statistics.

Step 2: Build Real-Time Monitor

Create monitoring script that polls transaction statistics every 5-10 seconds.

Step 3: Define Alert Thresholds

Set thresholds for long-running transactions, lock waits, and rollback rates.

Step 4: Implement Automated Actions

Auto-kill transactions exceeding thresholds or alert operators.

Step 5: Create Dashboards

Build Grafana dashboards for transaction metrics visualization.

Output Format

The command generates:

• monitoring/transaction_monitor.py - Real-time transaction monitoring daemon
• queries/transaction_analysis.sql - Transaction health diagnostic queries
• alerts/transaction_alerts.yml - Prometheus alerting rules
• dashboards/transaction_dashboard.json - Grafana dashboard configuration
• docs/transaction_runbook.md - Incident response procedures

Code Examples

Example 1: PostgreSQL Real-Time Transaction Monitor

# monitoring/postgres_transaction_monitor.py
import psycopg2
from psycopg2.extras import Dict Cursor
import time
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class TransactionInfo:
    """Represents an active transaction."""
    pid: int
    username: str
    database: str
    application_name: str
    client_addr: str
    state: str
    query: str
    transaction_start: datetime
    query_start: datetime
    wait_event: Optional[str]
    blocking_pids: List[int]

    def duration_seconds(self) -> float:
        return (datetime.now() - self.transaction_start).total_seconds()

    def to_dict(self) -> dict:
        result = asdict(self)
        result['transaction_start'] = self.transaction_start.isoformat()
        result['query_start'] = self.query_start.isoformat()
        result['duration_seconds'] = self.duration_seconds()
        return result

class PostgreSQLTransactionMonitor:
    """Monitor PostgreSQL transactions in real-time."""

    def __init__(
        self,
        connection_string: str,
        long_transaction_threshold: int = 30,
        check_interval: int = 10
    ):
        self.conn_string = connection_string
        self.long_transaction_threshold = long_transaction_threshold
        self.check_interval = check_interval
        self.stats = {
            'total_transactions': 0,
            'long_running_count': 0,
            'blocked_count': 0,
            'idle_in_transaction_count': 0
        }

    def connect(self):
        return psycopg2.connect(self.conn_string, cursor_factory=DictCursor)

    def get_active_transactions(self) -> List[TransactionInfo]:
        """Fetch all active transactions with blocking information."""
        query = """
        SELECT
            a.pid,
            a.usename,
            a.datname AS database,
            a.application_name,
            a.client_addr::text,
            a.state,
            a.query,
            a.xact_start AS transaction_start,
            a.query_start,
            a.wait_event,
            array_agg(b.pid) FILTER (WHERE b.pid IS NOT NULL) AS blocking_pids
        FROM pg_stat_activity a
        LEFT JOIN pg_stat_activity b ON b.pid = ANY(pg_blocking_pids(a.pid))
        WHERE a.pid != pg_backend_pid()
          AND a.state != 'idle'
          AND a.xact_start IS NOT NULL
        GROUP BY a.pid, a.usename, a.datname, a.application_name,
                 a.client_addr, a.state, a.query, a.xact_start,
                 a.query_start, a.wait_event
        ORDER BY a.xact_start;
        """

        conn = self.connect()
        try:
            with conn.cursor() as cur:
                cur.execute(query)
                rows = cur.fetchall()

                transactions = []
                for row in rows:
                    txn = TransactionInfo(
                        pid=row['pid'],
                        username=row['usename'],
                        database=row['database'],
                        application_name=row['application_name'] or 'unknown',
                        client_addr=row['client_addr'] or 'local',
                        state=row['state'],
                        query=row['query'][:200],  # Truncate long queries
                        transaction_start=row['transaction_start'],
                        query_start=row['query_start'],
                        wait_event=row['wait_event'],
                        blocking_pids=row['blocking_pids'] or []
                    )
                    transactions.append(txn)

                return transactions

        finally:
            conn.close()

    def find_long_running_transactions(
        self,
        transactions: List[TransactionInfo]
    ) -> List[TransactionInfo]:
        """Identify transactions exceeding threshold."""
        return [
            txn for txn in transactions
            if txn.duration_seconds() > self.long_transaction_threshold
        ]

    def find_blocked_transactions(
        self,
        transactions: List[TransactionInfo]
    ) -> List[TransactionInfo]:
        """Identify transactions waiting on locks."""
        return [
            txn for txn in transactions
            if txn.blocking_pids and len(txn.blocking_pids) > 0
        ]

    def find_idle_in_transaction(
        self,
        transactions: List[TransactionInfo]
    ) -> List[TransactionInfo]:
        """Find idle transactions holding locks."""
        return [
            txn for txn in transactions
            if txn.state == 'idle in transaction'
            and txn.duration_seconds() > 60
        ]

    def kill_transaction(self, pid: int, reason: str) -> bool:
        """Terminate a transaction by PID."""
        conn = self.connect()
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT pg_terminate_backend(%s)", (pid,))
                success = cur.fetchone()[0]

                if success:
                    logger.warning(f"Killed transaction PID {pid}: {reason}")
                else:
                    logger.error(f"Failed to kill transaction PID {pid}")

                return success

        finally:
            conn.close()

    def get_transaction_stats(self) -> Dict[str, any]:
        """Get overall transaction statistics."""
        conn = self.connect()
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    SELECT
                        (SELECT count(*) FROM pg_stat_activity WHERE state != 'idle') AS active_connections,
                        (SELECT count(*) FROM pg_stat_activity WHERE state = 'idle in transaction') AS idle_in_txn,
                        (SELECT sum(xact_commit) FROM pg_stat_database) AS total_commits,
                        (SELECT sum(xact_rollback) FROM pg_stat_database) AS total_rollbacks,
                        (SELECT sum(conflicts) FROM pg_stat_database_conflicts) AS conflicts
                """)

                row = cur.fetchone()

                total_txns = row['total_commits'] + row['total_rollbacks']
                rollback_rate = (row['total_rollbacks'] / total_txns * 100) if total_txns > 0 else 0

                return {
                    'active_connections': row['active_connections'],
                    'idle_in_transaction': row['idle_in_txn'],
                    'total_commits': row['total_commits'],
                    'total_rollbacks': row['total_rollbacks'],
                    'rollback_rate_percent': round(rollback_rate, 2),
                    'conflicts': row['conflicts']
                }

        finally:
            conn.close()

    def alert(self, severity: str, message: str, details: dict = None):
        """Send alert to monitoring system."""
        log_func = {
            'critical': logger.critical,
            'warning': logger.warning,
            'info': logger.info
        }.get(severity, logger.info)

        log_func(f"[{severity.upper()}] {message}")

        if details:
            logger.info(f"Details: {details}")

        # Implement webhook/email/PagerDuty integration here
        # Example: requests.post(webhook_url, json={'message': message, 'details': details})

    def run_monitoring_loop(self):
        """Main monitoring loop."""
        logger.info(f"Starting transaction monitoring (interval: {self.check_interval}s)")

        while True:
            try:
                # Fetch active transactions
                transactions = self.get_active_transactions()
                self.stats['total_transactions'] = len(transactions)

                # Check for long-running transactions
                long_running = self.find_long_running_transactions(transactions)
                if long_running:
                    self.stats['long_running_count'] = len(long_running)

                    for txn in long_running:
                        self.alert(
                            'warning',
                            f"Long-running transaction detected: PID {txn.pid}",
                            {
                                'duration': txn.duration_seconds(),
                                'database': txn.database,
                                'username': txn.username,
                                'query': txn.query
                            }
                        )

                        # Auto-kill if exceeds 5 minutes
                        if txn.duration_seconds() > 300:
                            self.kill_transaction(
                                txn.pid,
                                f"Exceeded 5 minute threshold ({txn.duration_seconds():.0f}s)"
                            )

                # Check for blocked transactions
                blocked = self.find_blocked_transactions(transactions)
                if blocked:
                    self.stats['blocked_count'] = len(blocked)

                    for txn in blocked:
                        self.alert(
                            'warning',
                            f"Blocked transaction: PID {txn.pid}",
                            {
                                'blocking_pids': txn.blocking_pids,
                                'wait_event': txn.wait_event,
                                'duration': txn.duration_seconds()
                            }
                        )

                # Check for idle in transaction
                idle_txns = self.find_idle_in_transaction(transactions)
                if idle_txns:
                    self.stats['idle_in_transaction_count'] = len(idle_txns)

                    for txn in idle_txns:
                        self.alert(
                            'warning',
                            f"Idle in transaction: PID {txn.pid}",
                            {
                                'duration': txn.duration_seconds(),
                                'application': txn.application_name
                            }
                        )

                        # Kill idle transactions after 10 minutes
                        if txn.duration_seconds() > 600:
                            self.kill_transaction(txn.pid, "Idle in transaction >10 minutes")

                # Get overall stats
                stats = self.get_transaction_stats()

                # Alert on high rollback rate
                if stats['rollback_rate_percent'] > 10:
                    self.alert(
                        'warning',
                        f"High transaction rollback rate: {stats['rollback_rate_percent']}%",
                        stats
                    )

                # Log periodic status
                logger.info(
                    f"Monitoring: {stats['active_connections']} active, "
                    f"{len(long_running)} long-running, "
                    f"{len(blocked)} blocked, "
                    f"{stats['rollback_rate_percent']}% rollback rate"
                )

                time.sleep(self.check_interval)

            except KeyboardInterrupt:
                logger.info("Monitoring stopped by user")
                break
            except Exception as e:
                logger.error(f"Monitoring error: {e}")
                time.sleep(self.check_interval)

# Usage
if __name__ == "__main__":
    monitor = PostgreSQLTransactionMonitor(
        connection_string="postgresql://monitor_user:password@localhost:5432/mydb",
        long_transaction_threshold=30,  # 30 seconds
        check_interval=10  # Check every 10 seconds
    )

    monitor.run_monitoring_loop()

Example 2: Transaction Analysis Queries

-- PostgreSQL transaction health diagnostic queries

-- 1. Long-running transactions
SELECT
    pid,
    usename,
    application_name,
    client_addr,
    NOW() - xact_start AS transaction_duration,
    NOW() - query_start AS query_duration,
    state,
    LEFT(query, 100) AS query_snippet
FROM pg_stat_activity
WHERE xact_start IS NOT NULL
  AND state != 'idle'
  AND pid != pg_backend_pid()
ORDER BY xact_start;

-- 2. Blocking tree (which transactions are blocking others)
WITH RECURSIVE blocking_tree AS (
    SELECT
        a.pid,
        a.usename,
        a.query AS blocked_query,
        NULL::integer AS blocking_pid,
        NULL::text AS blocking_query,
        1 AS level
    FROM pg_stat_activity a
    WHERE NOT EXISTS (
        SELECT 1 FROM pg_stat_activity b
        WHERE b.pid = ANY(pg_blocking_pids(a.pid))
    )
    AND a.pid IN (
        SELECT unnest(pg_blocking_pids(c.pid))
        FROM pg_stat_activity c
    )

    UNION ALL

    SELECT
        a.pid,
        a.usename,
        a.query,
        b.pid,
        b.query,
        bt.level + 1
    FROM blocking_tree bt
    JOIN pg_stat_activity a ON a.pid = ANY(
        SELECT unnest(pg_blocking_pids(x.pid))
        FROM pg_stat_activity x
        WHERE x.pid = bt.pid
    )
    JOIN pg_stat_activity b ON b.pid = ANY(pg_blocking_pids(a.pid))
)
SELECT
    level,
    pid,
    usename,
    blocking_pid,
    LEFT(blocked_query, 50) AS blocked_query,
    LEFT(blocking_query, 50) AS blocking_query
FROM blocking_tree
ORDER BY level, pid;

-- 3. Transaction rollback rate by database
SELECT
    datname,
    xact_commit AS commits,
    xact_rollback AS rollbacks,
    ROUND(100.0 * xact_rollback / NULLIF(xact_commit + xact_rollback, 0), 2) AS rollback_rate_percent
FROM pg_stat_database
WHERE datname NOT IN ('template0', 'template1', 'postgres')
ORDER BY rollback_rate_percent DESC;

-- 4. Idle in transaction connections
SELECT
    pid,
    usename,
    application_name,
    client_addr,
    NOW() - state_change AS idle_duration,
    state,
    query
FROM pg_stat_activity
WHERE state = 'idle in transaction'
  AND pid != pg_backend_pid()
ORDER BY state_change;

-- 5. Lock wait time by query
SELECT
    wait_event_type,
    wait_event,
    COUNT(*) AS waiting_count,
    array_agg(DISTINCT pid) AS waiting_pids
FROM pg_stat_activity
WHERE wait_event IS NOT NULL
  AND state = 'active'
GROUP BY wait_event_type, wait_event
ORDER BY waiting_count DESC;

Error Handling

Configuration Options

Monitoring Intervals

• check_interval: 5-10 seconds for real-time alerting
• long_transaction_threshold: 30-60 seconds (production), 300s (analytics)
• idle_in_transaction_timeout: 600 seconds (10 minutes)

Auto-Kill Thresholds

• Long-running OLTP: 60-300 seconds
• Long-running analytics: 3600 seconds (1 hour)
• Idle in transaction: 600 seconds (10 minutes)

Alert Thresholds

• Rollback rate: >5% warning, >10% critical
• Blocked transactions: >10 warning, >50 critical
• Active connections: >80% of max_connections

Best Practices

DO:

• Set statement_timeout in application connection strings
• Use connection pooling to limit total connections
• Implement transaction timeout in application code
• Monitor transaction throughput trends over time
• Kill idle in transaction connections automatically
• Track rollback reasons in application logs

DON'T:

• Leave transactions open while waiting for user input
• Hold locks during expensive operations (file I/O, network calls)
• Use long-running transactions in OLTP workloads
• Ignore idle in transaction connections (they hold locks)
• Set transaction timeouts too low (causes false positives)

Performance Considerations

• Monitoring adds <0.1% CPU overhead with 10-second intervals
• pg_stat_activity queries are lightweight (<1ms)
• Auto-killing transactions requires careful threshold tuning
• Historical metrics retention: 30 days (aggregated), 7 days (detailed)
• Consider read replicas for monitoring queries in high-load systems

Related Commands

• /database-deadlock-detector - Detailed deadlock analysis
• /database-health-monitor - Overall database health metrics
• /sql-query-optimizer - Optimize slow queries causing lock contention
• /database-connection-pooler - Manage connection pool sizing

Version History

• v1.0.0 (2024-10): Initial implementation with PostgreSQL real-time monitoring
• Planned v1.1.0: Add MySQL transaction monitoring and distributed transaction support