SQLite Litestream Incident Management

SQLite Litestream Incident

SQLite Litestream Incident Management Streaming Replication WAL S3 Backup Recovery Point-in-time RPO Single Binary Deploy

Litestream Configuration

# === Litestream Setup & Configuration ===

# Install Litestream
# wget https://github.com/benbjohnson/litestream/releases/download/v0.3.13/litestream-v0.3.13-linux-amd64.deb
# sudo dpkg -i litestream-v0.3.13-linux-amd64.deb
# # or: brew install litestream (macOS)

# litestream.yml Configuration
# dbs:
#   - path: /data/incidents.db
#     replicas:
#       - type: s3
#         bucket: my-backup-bucket
#         path: incidents
#         region: ap-southeast-1
#         retention: 168h  # 7 days
#         sync-interval: 1s
#
# # Alternative: File-based replica
#   - path: /data/incidents.db
#     replicas:
#       - type: file
#         path: /backup/incidents

# Start Replication
# litestream replicate -config litestream.yml

# Start with Application
# litestream replicate -config litestream.yml -exec "python app.py"

# Restore Database
# litestream restore -config litestream.yml -o /data/incidents.db

# Restore Point-in-time
# litestream restore -config litestream.yml -o /data/incidents.db \
#   -timestamp "2024-01-15T10:30:00Z"

# Check Snapshots
# litestream snapshots -config litestream.yml /data/incidents.db

from dataclasses import dataclass

@dataclass
class LitestreamConfig:
    setting: str
    value: str
    purpose: str
    impact: str

configs = [
    LitestreamConfig("replica.type",
        "s3 / gcs / abs / sftp / file",
        "ปลายทาง Backup",
        "S3 แนะนำ ราคาถูก Durable"),
    LitestreamConfig("retention",
        "168h (7 วัน)",
        "เก็บ Snapshot กี่วัน",
        "มาก = Restore ย้อนได้ไกล แต่ Cost เพิ่ม"),
    LitestreamConfig("sync-interval",
        "1s (Default 1 วินาที)",
        "ความถี่ส่ง WAL ไป Replica",
        "น้อย = RPO ต่ำ แต่ Request มาก"),
    LitestreamConfig("snapshot-interval",
        "1h (Default)",
        "ความถี่สร้าง Full Snapshot",
        "Restore เร็วขึ้น แต่ Storage เพิ่ม"),
    LitestreamConfig("validation-interval",
        "5m",
        "ตรวจ Checksum ความถูกต้อง",
        "ป้องกัน Silent Corruption"),
]

print("=== Litestream Configs ===")
for c in configs:
    print(f"  [{c.setting}] = {c.value}")
    print(f"    Purpose: {c.purpose}")
    print(f"    Impact: {c.impact}")

Incident Tracking Schema

# === Incident Management Database Schema ===

# CREATE TABLE incidents (
#     id INTEGER PRIMARY KEY AUTOINCREMENT,
#     incident_id TEXT UNIQUE NOT NULL,
#     title TEXT NOT NULL,
#     description TEXT,
#     severity TEXT CHECK(severity IN ('critical','high','medium','low')),
#     status TEXT CHECK(status IN ('open','investigating','mitigating','resolved','closed')),
#     reported_by TEXT,
#     assigned_to TEXT,
#     reported_at DATETIME DEFAULT CURRENT_TIMESTAMP,
#     resolved_at DATETIME,
#     root_cause TEXT,
#     impact TEXT,
#     affected_services TEXT,
#     mttd_minutes REAL,
#     mttr_minutes REAL
# );
#
# CREATE TABLE incident_updates (
#     id INTEGER PRIMARY KEY AUTOINCREMENT,
#     incident_id TEXT REFERENCES incidents(incident_id),
#     update_text TEXT NOT NULL,
#     updated_by TEXT,
#     new_status TEXT,
#     updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
# );
#
# CREATE INDEX idx_incidents_status ON incidents(status);
# CREATE INDEX idx_incidents_severity ON incidents(severity);
# CREATE INDEX idx_updates_incident ON incident_updates(incident_id);

@dataclass
class IncidentRecord:
    incident_id: str
    title: str
    severity: str
    status: str
    mttd: float
    mttr: float

sample_incidents = [
    IncidentRecord("INC-2024-001",
        "API Gateway 502 Errors Spike",
        "critical", "resolved", 3.0, 45.0),
    IncidentRecord("INC-2024-002",
        "Database Connection Pool Exhausted",
        "high", "resolved", 5.0, 30.0),
    IncidentRecord("INC-2024-003",
        "SSL Certificate Expired on CDN",
        "critical", "resolved", 1.0, 15.0),
    IncidentRecord("INC-2024-004",
        "Memory Leak in Worker Service",
        "medium", "investigating", 120.0, 0),
    IncidentRecord("INC-2024-005",
        "Slow Query on Reports Dashboard",
        "low", "open", 0, 0),
]

print("=== Incident Records ===")
for i in sample_incidents:
    print(f"  [{i.incident_id}] {i.title}")
    print(f"    Severity: {i.severity} | Status: {i.status}")
    print(f"    MTTD: {i.mttd} min | MTTR: {i.mttr} min")

Recovery & Operations

# === Disaster Recovery Procedures ===

@dataclass
class RecoveryScenario:
    scenario: str
    steps: str
    command: str
    rto: str
    rpo: str

scenarios = [
    RecoveryScenario("Server Crash (Total Loss)",
        "1. Provision new server\n"
        "2. Install Litestream + App\n"
        "3. Restore DB from S3\n"
        "4. Start Litestream + App",
        "litestream restore -config litestream.yml -o /data/incidents.db",
        "5-15 นาที",
        "< 10 วินาที (Last WAL sync)"),
    RecoveryScenario("Data Corruption",
        "1. Stop Application\n"
        "2. Restore to point before corruption\n"
        "3. Verify data integrity\n"
        "4. Restart App + Litestream",
        "litestream restore -o /data/incidents.db -timestamp '2024-01-15T10:00:00Z'",
        "5 นาที",
        "ตาม timestamp ที่เลือก"),
    RecoveryScenario("Accidental DELETE",
        "1. Stop Application\n"
        "2. Restore latest snapshot\n"
        "3. Compare with current DB\n"
        "4. Recover deleted records",
        "litestream restore -o /tmp/backup.db && sqlite3 /tmp/backup.db 'SELECT ...'",
        "5 นาที",
        "< 10 วินาที"),
    RecoveryScenario("S3 Bucket Issue",
        "1. Check S3 status\n"
        "2. Use local WAL files if available\n"
        "3. Switch to backup replica\n"
        "4. Fix S3 and re-sync",
        "litestream restore -replica file -o /data/incidents.db",
        "5-10 นาที",
        "ตาม Local WAL"),
]

print("=== Recovery Scenarios ===")
for r in scenarios:
    print(f"\n  [{r.scenario}]")
    print(f"    Steps: {r.steps}")
    print(f"    Command: {r.command}")
    print(f"    RTO: {r.rto} | RPO: {r.rpo}")

เคล็ดลับ

• S3: ใช้ S3 เป็น Replica หลัก ราคาถูก Durable 99.999999999%
• Test: ทดสอบ Restore ทุกเดือน ตรวจว่า Backup ใช้ได้จริง
• WAL: เปิด WAL Mode ใน SQLite (PRAGMA journal_mode=WAL)
• Systemd: ใช้ Systemd Service สำหรับ Production Auto-restart
• Monitor: ตรวจ Litestream Logs หา Replication Error

Best Practices สำหรับนักพัฒนา

การเขียนโค้ดที่ดีไม่ใช่แค่ทำให้โปรแกรมทำงานได้ แต่ต้องเขียนให้อ่านง่าย ดูแลรักษาง่าย และ Scale ได้ หลัก SOLID Principles เป็นพื้นฐานสำคัญที่นักพัฒนาทุกู้คืนควรเข้าใจ ได้แก่ Single Responsibility ที่แต่ละ Class ทำหน้าที่เดียว Open-Closed ที่เปิดให้ขยายแต่ปิดการแก้ไข Liskov Substitution ที่ Subclass ต้องใช้แทน Parent ได้ Interface Segregation ที่แยก Interface ให้เล็ก และ Dependency Inversion ที่พึ่งพา Abstraction ไม่ใช่ Implementation

เรื่อง Testing ก็ขาดไม่ได้ ควรเขียน Unit Test ครอบคลุมอย่างน้อย 80% ของ Code Base ใช้ Integration Test ทดสอบการทำงานร่วมกันของ Module ต่างๆ และ E2E Test สำหรับ Critical User Flow เครื่องมือยอดนิยมเช่น Jest, Pytest, JUnit ช่วยให้การเขียน Test เป็นเรื่องง่าย

เรื่อง Version Control ด้วย Git ใช้ Branch Strategy ที่เหมาะกับทีม เช่น Git Flow สำหรับโปรเจคใหญ่ หรือ Trunk-Based Development สำหรับทีมที่ Deploy บ่อย ทำ Code Review ทุก Pull Request และใช้ CI/CD Pipeline ทำ Automated Testing และ Deployment

สรุป

SQLite Litestream Incident Management WAL S3 Replication Backup Restore Point-in-time RPO RTO Single Binary Deploy Cost Effective Production