tools-only 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:
- Database monitoring permissions (pg_monitor role or PROCESS privilege)
- Access to pg_stat_activity (PostgreSQL) or performance_schema (MySQL)
- Alerting infrastructure (Slack, PagerDuty, email)
- Monitoring data retention strategy (metrics database or time-series DB)
- 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 daemonqueries/transaction_analysis.sql- Transaction health diagnostic queriesalerts/transaction_alerts.yml- Prometheus alerting rulesdashboards/transaction_dashboard.json- Grafana dashboard configurationdocs/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
| Error | Cause | Solution |
|---|---|---|
| "Permission denied for pg_stat_activity" | Insufficient monitoring privileges | Grant pg_monitor role or SELECT on pg_stat_activity |
| "Cannot terminate backend" | Trying to kill superuser connection | Use pg_cancel_backend or kill from OS level |
| "Connection pool exhausted" | Too many idle connections | Kill idle in transaction connections, increase pool size |
| "High rollback rate" | Application errors or constraint violations | Review application logs and fix bugs |
| "Lock wait timeout exceeded" | Deadlock or very long lock hold | Analyze blocking queries, implement timeouts |
Configuration Options
Monitoring Intervals
check_interval: 5-10 seconds for real-time alertinglong_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