Redis Memory Diagnosis Script — analyze memory usage, key prefix distribution, TTL coverage, and big keys on any Redis instance

redis_memory_diagnosis.py

#!/usr/bin/env python3
"""
Redis Memory Diagnosis Script

Connects to a Redis instance and produces a comprehensive memory analysis:
- Memory summary (used, peak, RSS, fragmentation, eviction stats)
- Replication topology (master/replica, connected replicas)
- Keyspace overview (key counts, TTL coverage per DB)
- Key prefix distribution (what categories of keys exist and how much space they use)
- TTL analysis per prefix (which prefixes lack TTLs)
- Estimated memory by prefix (extrapolated from sample to full keyspace)
- Top large keys by memory usage

Designed for Todoist ElastiCache instances but works on any Redis server.

Usage:
    python redis_memory_diagnosis.py <host:port>
    python redis_memory_diagnosis.py todoist-redis16.iahb8b.ng.0001.use1.cache.amazonaws.com:6379

Options:
    --db DB              Database number to analyze (default: 0)
    --scan-count N       Keys to sample for prefix analysis (default: 50000)
    --big-key-threshold  Bytes threshold for "big key" reporting (default: 102400 / 100KB)

Examples:
    python redis_memory_diagnosis.py redis16.example.com:6379
    python redis_memory_diagnosis.py redis16.example.com:6379 --scan-count 200000
    python redis_memory_diagnosis.py redis16.example.com:6379 --db 1
    python redis_memory_diagnosis.py redis16.example.com:6379 --big-key-threshold 51200
"""

from __future__ import annotations

import argparse
import sys
import time
from collections import defaultdict

from redis import Redis


def connect(host: str, port: int, db: int = 0) -> Redis:
    # decode_responses=False because some Redis instances store binary keys
    return Redis(
        host=host,
        port=port,
        db=db,
        socket_connect_timeout=5,
        socket_timeout=30,
        decode_responses=False,
    )


def safe_decode(value: bytes | str) -> str:
    """Decode bytes to string, replacing invalid UTF-8 sequences."""
    if isinstance(value, str):
        return value
    return value.decode("utf-8", errors="replace")


def fmt_bytes(n: float) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if abs(n) < 1024:
            return f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} TB"


def fmt_number(n: int) -> str:
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f}M"
    if n >= 1_000:
        return f"{n / 1_000:.1f}K"
    return str(n)


def section_header(title: str) -> None:
    print(f"\n{'=' * 70}")
    print(f"  {title}")
    print(f"{'=' * 70}")


def extract_prefix(key: str) -> str:
    """Extract a meaningful prefix from a Redis key.

    Tries common separators and takes the first 1-2 segments to group
    keys by purpose. Falls back to the raw key for short keys or
    a truncated version for long ones.
    """
    for sep in (":", "/", "."):
        if sep in key:
            parts = key.split(sep)
            if len(parts) >= 2:
                second = parts[1]
                # If second part looks like an ID, just use first part
                if second.isdigit() or (len(second) > 20 and not second.isalpha()):
                    return parts[0] + sep + "*"
                if len(parts) > 2:
                    return parts[0] + sep + parts[1] + sep + "*"
                return parts[0] + sep + parts[1]
            return parts[0] + sep + "*"
    return key if len(key) <= 30 else key[:30] + "..."


def print_memory_summary(r: Redis) -> None:
    section_header("MEMORY SUMMARY")
    mem = r.info("memory")
    stats = r.info("stats")
    keyspace = r.info("keyspace")
    repl = r.info("replication")
    clients = r.info("clients")

    used = mem.get("used_memory", 0)
    maxmem = mem.get("maxmemory", 0)
    pct = f"{used / maxmem * 100:.1f}%" if maxmem else "N/A (no maxmemory set)"

    print(f"  Used memory:         {fmt_bytes(used)} / {fmt_bytes(maxmem)} ({pct})")
    print(f"  Peak memory:         {fmt_bytes(mem.get('used_memory_peak', 0))}")
    print(f"  RSS:                 {fmt_bytes(mem.get('used_memory_rss', 0))}")
    print(f"  Dataset:             {fmt_bytes(mem.get('used_memory_dataset', 0))}")
    print(f"  Overhead:            {fmt_bytes(mem.get('used_memory_overhead', 0))}")
    print(f"  Repl backlog:        {fmt_bytes(mem.get('mem_replication_backlog', 0))}")
    print(f"  Clients (normal):    {fmt_bytes(mem.get('mem_clients_normal', 0))}")
    print(f"  Clients (replicas):  {fmt_bytes(mem.get('mem_clients_slaves', 0))}")
    print(f"  AOF buffer:          {fmt_bytes(mem.get('mem_aof_buffer', 0))}")
    print(f"  Fragmentation ratio: {mem.get('mem_fragmentation_ratio', 'N/A')}")
    print(f"  Allocator:           {mem.get('mem_allocator', 'N/A')}")
    print()

    # Replication
    role = repl.get("role", "unknown")
    print(f"  Role:                {role}")
    if role == "master":
        print(f"  Connected replicas:  {repl.get('connected_slaves', 0)}")
    elif role == "slave":
        print(f"  Master:              {repl.get('master_host', '?')}:{repl.get('master_port', '?')}")
        print(f"  Link status:         {repl.get('master_link_status', '?')}")
    print()

    # Stats
    print(f"  Evicted keys:        {fmt_number(stats.get('evicted_keys', 0))}")
    print(f"  Expired keys:        {fmt_number(stats.get('expired_keys', 0))}")
    print(f"  Keyspace hits:       {fmt_number(stats.get('keyspace_hits', 0))}")
    print(f"  Keyspace misses:     {fmt_number(stats.get('keyspace_misses', 0))}")
    print(f"  Connected clients:   {clients.get('connected_clients', '?')}")
    print()

    # Keyspace
    if not keyspace:
        print("  (no databases with keys)")
    for db_name, db_stats in sorted(keyspace.items()):
        total = db_stats["keys"]
        with_ttl = db_stats["expires"]
        no_ttl = total - with_ttl
        avg_ttl_ms = db_stats.get("avg_ttl", 0)
        avg_ttl_human = f"{avg_ttl_ms / 1000 / 3600:.1f} hours" if avg_ttl_ms else "N/A"
        no_ttl_pct = f"{no_ttl / total * 100:.1f}%" if total else "0%"
        print(
            f"  {db_name}: {fmt_number(total)} keys total, "
            f"{fmt_number(with_ttl)} with TTL, "
            f"{fmt_number(no_ttl)} WITHOUT TTL ({no_ttl_pct}), "
            f"avg TTL: {avg_ttl_human}"
        )


def analyze_key_prefixes(
    r: Redis, db: int, scan_limit: int, big_key_threshold: int
) -> None:
    section_header(f"KEY PREFIX ANALYSIS (DB {db}, sampling {fmt_number(scan_limit)} keys)")

    prefix_stats: dict[str, dict] = defaultdict(
        lambda: {
            "count": 0,
            "with_ttl": 0,
            "without_ttl": 0,
            "total_memory": 0,
            "types": defaultdict(int),
        }
    )
    big_keys: list[tuple[str, str, int, int]] = []

    scanned = 0
    cursor = 0
    start_time = time.time()
    last_progress = 0

    while scanned < scan_limit:
        cursor, keys = r.scan(cursor=cursor, count=500)
        if not keys:
            if cursor == 0:
                break
            continue

        pipe = r.pipeline(transaction=False)
        for key in keys:
            pipe.type(key)
            pipe.ttl(key)
            pipe.memory_usage(key, samples=0)
        results = pipe.execute()

        for i, raw_key in enumerate(keys):
            key = safe_decode(raw_key)
            key_type = safe_decode(results[i * 3])
            ttl = results[i * 3 + 1]
            mem = results[i * 3 + 2] or 0

            prefix = extract_prefix(key)
            stats = prefix_stats[prefix]
            stats["count"] += 1
            stats["total_memory"] += mem
            stats["types"][key_type] += 1

            if ttl == -1:
                stats["without_ttl"] += 1
            else:
                stats["with_ttl"] += 1

            if mem >= big_key_threshold:
                big_keys.append((key, key_type, mem, ttl))

            scanned += 1
            if scanned >= scan_limit:
                break

        if scanned - last_progress >= 10000:
            elapsed = time.time() - start_time
            rate = scanned / elapsed if elapsed > 0 else 0
            print(f"  ... scanned {fmt_number(scanned)} keys ({rate:.0f} keys/sec)")
            last_progress = scanned

        if cursor == 0:
            break

    elapsed = time.time() - start_time
    print(f"  Scanned {fmt_number(scanned)} keys in {elapsed:.1f}s\n")

    if not prefix_stats:
        print("  No keys found in sample.")
        return

    sorted_prefixes = sorted(
        prefix_stats.items(), key=lambda x: x[1]["total_memory"], reverse=True
    )

    # Top prefixes by memory
    print(
        f"  {'Prefix':<45} {'Keys':>10} {'Memory':>12} "
        f"{'No TTL':>10} {'No TTL%':>8} {'Types'}"
    )
    print(f"  {'-' * 45} {'-' * 10} {'-' * 12} {'-' * 10} {'-' * 8} {'-' * 20}")

    for prefix, stats in sorted_prefixes[:40]:
        no_ttl_pct = (
            f"{stats['without_ttl'] / stats['count'] * 100:.0f}%"
            if stats["count"]
            else "0%"
        )
        types_str = ", ".join(
            f"{t}:{c}"
            for t, c in sorted(stats["types"].items(), key=lambda x: -x[1])
        )
        print(
            f"  {prefix:<45} {stats['count']:>10} "
            f"{fmt_bytes(stats['total_memory']):>12} "
            f"{stats['without_ttl']:>10} {no_ttl_pct:>8} {types_str}"
        )

    # Top prefixes by no-TTL count
    sorted_by_no_ttl = sorted(
        prefix_stats.items(), key=lambda x: x[1]["without_ttl"], reverse=True
    )
    has_no_ttl = any(s["without_ttl"] > 0 for _, s in sorted_by_no_ttl)

    if has_no_ttl:
        section_header(f"TOP PREFIXES BY NO-TTL KEY COUNT (DB {db})")
        print(
            f"  {'Prefix':<45} {'No TTL':>10} {'Total':>10} "
            f"{'No TTL%':>8} {'Memory':>12}"
        )
        print(f"  {'-' * 45} {'-' * 10} {'-' * 10} {'-' * 8} {'-' * 12}")

        for prefix, stats in sorted_by_no_ttl[:30]:
            if stats["without_ttl"] == 0:
                break
            no_ttl_pct = f"{stats['without_ttl'] / stats['count'] * 100:.0f}%"
            print(
                f"  {prefix:<45} {stats['without_ttl']:>10} "
                f"{stats['count']:>10} "
                f"{no_ttl_pct:>8} {fmt_bytes(stats['total_memory']):>12}"
            )

    # Extrapolated totals
    keyspace = r.info("keyspace")
    db_key = f"db{db}"
    if db_key in keyspace and scanned > 0:
        total_keys = keyspace[db_key]["keys"]
        scale = total_keys / scanned

        section_header(
            f"ESTIMATED MEMORY BY PREFIX (extrapolated to {fmt_number(total_keys)} keys)"
        )
        print(
            f"  Sample: {fmt_number(scanned)} keys, "
            f"Total: {fmt_number(total_keys)} keys, "
            f"Scale factor: {scale:.1f}x\n"
        )
        print(
            f"  {'Prefix':<45} {'Est. Keys':>12} "
            f"{'Est. Memory':>14} {'Est. No-TTL':>12}"
        )
        print(f"  {'-' * 45} {'-' * 12} {'-' * 14} {'-' * 12}")

        for prefix, stats in sorted_prefixes[:25]:
            est_keys = int(stats["count"] * scale)
            est_mem = stats["total_memory"] * scale
            est_no_ttl = int(stats["without_ttl"] * scale)
            print(
                f"  {prefix:<45} {fmt_number(est_keys):>12} "
                f"{fmt_bytes(est_mem):>14} {fmt_number(est_no_ttl):>12}"
            )

    # Big keys
    if big_keys:
        section_header(f"BIG KEYS (>{fmt_bytes(big_key_threshold)}) FOUND IN DB {db}")
        big_keys.sort(key=lambda x: x[2], reverse=True)
        print(
            f"  {'Key':<60} {'Type':<8} {'Memory':>12} {'TTL':>10}"
        )
        print(f"  {'-' * 60} {'-' * 8} {'-' * 12} {'-' * 10}")
        for key, ktype, mem, ttl in big_keys[:50]:
            display_key = key if len(key) <= 58 else key[:55] + "..."
            ttl_str = "NO TTL" if ttl == -1 else f"{ttl}s"
            print(
                f"  {display_key:<60} {ktype:<8} "
                f"{fmt_bytes(mem):>12} {ttl_str:>10}"
            )
    else:
        print(
            f"\n  No keys >{fmt_bytes(big_key_threshold)} found in DB {db} sample."
        )


def parse_endpoint(endpoint: str) -> tuple[str, int]:
    if ":" in endpoint:
        h, p = endpoint.rsplit(":", 1)
        return h, int(p)
    return endpoint, 6379


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Analyze memory usage of a Redis instance.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument(
        "endpoint",
        help="Redis endpoint as host:port (e.g. redis.example.com:6379)",
    )
    parser.add_argument(
        "--db",
        type=int,
        default=0,
        help="Database number to analyze (default: 0)",
    )
    parser.add_argument(
        "--scan-count",
        type=int,
        default=50000,
        help="Number of keys to sample for prefix analysis (default: 50000)",
    )
    parser.add_argument(
        "--big-key-threshold",
        type=int,
        default=100 * 1024,
        help="Bytes threshold for big key reporting (default: 102400 / 100KB)",
    )
    args = parser.parse_args()

    host, port = parse_endpoint(args.endpoint)

    print(f"Connecting to {host}:{port} (db{args.db})...")
    r = connect(host, port, db=args.db)

    try:
        r.ping()
    except Exception as e:
        print(f"ERROR: Cannot connect to {host}:{port}: {e}")
        sys.exit(1)

    print(
        f"Connected. Scan limit: {fmt_number(args.scan_count)} keys, "
        f"big key threshold: {fmt_bytes(args.big_key_threshold)}"
    )

    print_memory_summary(r)
    analyze_key_prefixes(
        r,
        db=args.db,
        scan_limit=args.scan_count,
        big_key_threshold=args.big_key_threshold,
    )

    section_header("DONE")
    print("  Next steps based on results:")
    print("    1. Identify prefixes with high no-TTL counts — these are memory growth drivers")
    print("    2. Trace prefixes back to application code to understand what's writing them")
    print("    3. Check the eviction policy in AWS console (Parameter Group)")
    print("    4. If evictions are happening, the instance may need scaling up")
    print("    5. If no-TTL keys dominate, add TTLs or run cleanup scripts")
    print()

    r.close()


if __name__ == "__main__":
    main()