#!/usr/bin/env python3 """ keystats.py - Redis keyspace analysis tool Unlike redis-cli --bigkeys/--memkeys, this scans ALL keys (not sampled). Usage: ./keystats.py memkeys -u redis://localhost --mem-limit 100mb > bigkeys.csv ./keystats.py slots -u redis://localhost --shards 140 > slots.csv Requirements: - Python 3.7+ - redis-py (pip install redis) """ import argparse import sys from collections import defaultdict try: import redis except ImportError: print("Error: 'redis' module not found. Install it with:", file=sys.stderr) print(" pip3 install redis --break-system-packages", file=sys.stderr) exit(1) # CRC16 table for Redis slot calculation CRC16_TABLE = [ 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7, 0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef, 0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6, 0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de, 0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485, 0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d, 0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4, 0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc, 0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823, 0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b, 0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12, 0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a, 0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41, 0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49, 0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70, 0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78, 0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f, 0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e, 0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256, 0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d, 0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, 0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c, 0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634, 0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab, 0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3, 0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a, 0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92, 0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9, 0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1, 0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8, 0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0, ] def crc16(data: bytes) -> int: crc = 0 for byte in data: crc = ((crc << 8) & 0xFFFF) ^ CRC16_TABLE[((crc >> 8) ^ byte) & 0xFF] return crc def key_slot(key: str) -> int: """Calculate Redis cluster slot for a key (0-16383).""" key_bytes = key.encode('utf-8') if isinstance(key, str) else key start = key_bytes.find(b'{') if start != -1: end = key_bytes.find(b'}', start + 1) if end != -1 and end != start + 1: key_bytes = key_bytes[start + 1:end] return crc16(key_bytes) & 16383 def format_size(size_bytes: int) -> str: for unit in ['B', 'KB', 'MB', 'GB']: if abs(size_bytes) < 1024: return f"{size_bytes:.2f} {unit}" size_bytes /= 1024 return f"{size_bytes:.2f} TB" def parse_size(size_str: str) -> int: """Parse size string like '100mb' or '1gb' to bytes.""" size_str = size_str.lower().strip() # Check longer suffixes first to avoid 'b' matching before 'mb' multipliers = [('tb', 1024**4), ('gb', 1024**3), ('mb', 1024**2), ('kb', 1024), ('b', 1)] for suffix, mult in multipliers: if size_str.endswith(suffix): return int(float(size_str[:-len(suffix)]) * mult) return int(size_str) def scan_big_keys(r: redis.Redis, min_size_bytes: int, scan_count: int = 1000, num_shards: int = None, collect_keys: bool = True) -> tuple[list[dict], dict]: """Scan all keys and return those above the threshold, plus shard totals if requested. Args: r: Redis connection min_size_bytes: Minimum size threshold for collecting keys scan_count: SCAN batch size num_shards: Number of shards for slot distribution analysis collect_keys: If False, skip collecting big keys (saves memory for slots-only analysis) """ big_keys = [] if collect_keys else None shard_totals = defaultdict(int) if num_shards else None slots_per_shard = 16384 // num_shards if num_shards else None cursor = 0 total_scanned = 0 total_bytes = 0 if collect_keys: print(f"Scanning keys (threshold: {format_size(min_size_bytes)})...", file=sys.stderr, flush=True) else: print("Scanning keys for slot distribution...", file=sys.stderr, flush=True) while True: cursor, keys = r.scan(cursor=cursor, count=scan_count) total_scanned += len(keys) if total_scanned % 10000 == 0: if collect_keys: print(f" Scanned {total_scanned} keys, found {len(big_keys)} big keys, total {format_size(total_bytes)}...", file=sys.stderr, flush=True) else: print(f" Scanned {total_scanned} keys, total {format_size(total_bytes)}...", file=sys.stderr, flush=True) # Pipeline MEMORY USAGE commands for efficiency pipe = r.pipeline(transaction=False) for key in keys: pipe.memory_usage(key) sizes = pipe.execute() for key, size in zip(keys, sizes): if size is None: continue key_str = key.decode('utf-8') if isinstance(key, bytes) else key total_bytes += size # Track shard totals for ALL keys if num_shards: slot = key_slot(key_str) # Clamp shard to valid range (handles case where 16384 % num_shards != 0) shard = min(slot // slots_per_shard, num_shards - 1) shard_totals[shard] += size # Only collect big keys for output (if enabled) if collect_keys and size >= min_size_bytes: big_keys.append({'key': key_str, 'size': size}) if cursor == 0: break if collect_keys: print(f"Done. Scanned {total_scanned} keys, total {format_size(total_bytes)}, found {len(big_keys)} above threshold.", file=sys.stderr, flush=True) else: print(f"Done. Scanned {total_scanned} keys, total {format_size(total_bytes)}.", file=sys.stderr, flush=True) return big_keys, shard_totals def add_connection_args(parser): """Add common Redis connection arguments to a parser.""" parser.add_argument('-u', '--uri', default=None, help='Redis URI') parser.add_argument('-h', '--host', default='localhost', help='Redis host') parser.add_argument('-p', '--port', type=int, default=6379, help='Redis port') parser.add_argument('-a', '--password', default=None, help='Password') parser.add_argument('--user', default=None, help='ACL username') parser.add_argument('--askpass', action='store_true', help='Prompt for password') parser.add_argument('-n', '--db', type=int, default=0, help='Database number') parser.add_argument('--timeout', type=float, default=None, help='Timeout in seconds') parser.add_argument('--tls', action='store_true', help='Use TLS') parser.add_argument('--cacert', default=None, help='CA cert file') parser.add_argument('--cert', default=None, help='Client cert file') parser.add_argument('--key', default=None, help='Client key file') parser.add_argument('--insecure', action='store_true', help='Skip cert verification') parser.add_argument('--scan-count', type=int, default=1000, help='SCAN batch size') def connect(args) -> redis.Redis: """Create Redis connection from args.""" import getpass password = args.password if args.askpass: password = getpass.getpass("Password: ") if args.uri: # Warn if conflicting args are provided (they will be ignored) ignored = [] if args.password or args.askpass: ignored.append('password/askpass') if args.user: ignored.append('user') if args.host != 'localhost': ignored.append('host') if args.port != 6379: ignored.append('port') if args.db != 0: ignored.append('db') if args.tls: ignored.append('tls') if args.cacert or args.cert or args.key: ignored.append('cert options') if args.insecure: ignored.append('insecure') if ignored: print(f"Warning: --uri provided, ignoring: {', '.join(ignored)}", file=sys.stderr) return redis.from_url(args.uri, socket_timeout=args.timeout) else: return redis.Redis( host=args.host, port=args.port, username=args.user, password=password, db=args.db, ssl=args.tls, ssl_cert_reqs='none' if args.insecure else None, ssl_ca_certs=args.cacert, ssl_certfile=args.cert, ssl_keyfile=args.key, socket_timeout=args.timeout, ) def cmd_memkeys(args): """Find all keys above a memory threshold.""" r = connect(args) try: r.ping() except redis.ConnectionError as e: print(f"Connection failed: {e}", file=sys.stderr) return 1 min_size = parse_size(args.mem_limit) big_keys, _ = scan_big_keys(r, min_size, args.scan_count, num_shards=None) # CSV output print("key,size_bytes") for k in sorted(big_keys, key=lambda x: x['size'], reverse=True): key = k['key'].replace('"', '""') print(f'"{key}",{k["size"]}') return 0 def cmd_slots(args): """Analyze memory distribution across slot ranges.""" # Validate shards if args.shards < 1 or args.shards > 16384: print(f"Error: --shards must be between 1 and 16384 (got {args.shards})", file=sys.stderr) return 1 r = connect(args) try: r.ping() except redis.ConnectionError as e: print(f"Connection failed: {e}", file=sys.stderr) return 1 # Use collect_keys=False to avoid building a huge list we don't need _, shard_totals = scan_big_keys(r, min_size_bytes=0, scan_count=args.scan_count, num_shards=args.shards, collect_keys=False) slots_per_shard = 16384 // args.shards # CSV output print("shard,slot_start,slot_end,size_bytes") for shard in range(args.shards): slot_start = shard * slots_per_shard # Last shard gets all remaining slots (handles 16384 % shards != 0) slot_end = 16383 if shard == args.shards - 1 else slot_start + slots_per_shard - 1 print(f'{shard},{slot_start},{slot_end},{shard_totals.get(shard, 0)}') return 0 def main(): parser = argparse.ArgumentParser( description='Redis key analysis tools', formatter_class=argparse.RawDescriptionHelpFormatter, add_help=False, ) parser.add_argument('--help', action='help', help='Show this help message and exit') subparsers = parser.add_subparsers(dest='command', help='Available commands') # memkeys subcommand memkeys_parser = subparsers.add_parser( 'memkeys', help='Find all keys above a memory threshold', description='Scan Redis for ALL keys above a size threshold (complete, not sampled)', add_help=False, ) memkeys_parser.add_argument('--help', action='help', help='Show this help message and exit') add_connection_args(memkeys_parser) memkeys_parser.add_argument('--mem-limit', default='100mb', help='Min key size (default: 100mb)') memkeys_parser.set_defaults(func=cmd_memkeys) # slots subcommand slots_parser = subparsers.add_parser( 'slots', help='Analyze memory distribution across slot ranges', description='Calculate total memory per slot range (shard)', add_help=False, ) slots_parser.add_argument('--help', action='help', help='Show this help message and exit') add_connection_args(slots_parser) slots_parser.add_argument('--shards', type=int, default=100, help='Number of shards (default: 100)') slots_parser.set_defaults(func=cmd_slots) args = parser.parse_args() if not args.command: parser.print_help() return 1 return args.func(args) if __name__ == '__main__': exit(main())