#!/usr/bin/env python3 """ ElastiCache Migration Preflight Check Connects to a source Redis/Valkey instance and evaluates migration readiness: - Engine version compatibility - Module detection (RedisJSON, RediSearch, RedisTimeSeries, etc.) - Key count and memory usage - Persistence configuration - Cluster mode detection - Unsupported command sampling Usage: # Check a source Redis instance (plain) python migration_preflight.py --host redis.example.com --port 6379 # With TLS python migration_preflight.py --host redis.example.com --port 6379 --tls # With auth python migration_preflight.py --host redis.example.com --password mypassword # JSON output python migration_preflight.py --host redis.example.com --json # Via tunnel (local port forward) python migration_preflight.py --host localhost --port 6379 Requires: valkey (pip install valkey) """ import argparse import json import sys # --------------------------------------------------------------------------- # Dependency check -- fail early with an actionable message instead of a # raw ImportError traceback. # --------------------------------------------------------------------------- try: import valkey as redis except ImportError: print( "Error: the 'valkey' package is required for this script.\n" "Install with:\n" " pip install valkey" ) sys.exit(1) # Commands not available on ElastiCache (restricted or modified) ELASTICACHE_RESTRICTED_COMMANDS = { "BGSAVE", "BGREWRITEAOF", "CONFIG", "DEBUG", "MIGRATE", "SAVE", "SHUTDOWN", "SLAVEOF", "REPLICAOF", "SYNC", } # Known Redis modules and their ElastiCache availability MODULE_COMPATIBILITY = { "search": {"name": "RediSearch", "elasticache": False, "note": "Not available on ElastiCache. Valkey 8.2 provides native vector search (FT.* commands) on node-based clusters."}, "ReJSON": {"name": "RedisJSON", "elasticache": False, "note": "The ReJSON module cannot be loaded on ElastiCache. However, ElastiCache provides native JSON support (JSON.* commands) in Valkey 7.2+ and Redis OSS 6.2.6+. Verify command compatibility before migration."}, "timeseries": {"name": "RedisTimeSeries", "elasticache": False, "note": "Not available on ElastiCache. Consider CloudWatch or Timestream."}, "bf": {"name": "RedisBloom", "elasticache": False, "note": "No (module) / Yes (native BF.* commands in Valkey 8.1+)."}, "graph": {"name": "RedisGraph", "elasticache": False, "note": "Deprecated. Consider Neptune or OpenSearch."}, "ai": {"name": "RedisAI", "elasticache": False, "note": "Not available on ElastiCache. Use SageMaker for ML inference."}, } def run_preflight(host, port, password=None, use_tls=False, username=None): """Run all preflight checks against a source Redis/Valkey instance.""" findings = [] info = {"host": host, "port": port} try: r = redis.Redis( host=host, port=port, password=password, username=username, ssl=use_tls, decode_responses=True, socket_timeout=10, socket_connect_timeout=10, ) r.ping() findings.append({"check": "Connectivity", "status": "PASS", "detail": f"Connected to {host}:{port}"}) except (redis.ConnectionError, redis.TimeoutError) as e: return {"error": f"Cannot connect to {host}:{port}: {e}", "info": info, "findings": [ {"check": "Connectivity", "status": "FAIL", "detail": str(e)} ]} except redis.AuthenticationError as e: return {"error": f"Authentication failed: {e}", "info": info, "findings": [ {"check": "Connectivity", "status": "FAIL", "detail": f"Auth failed: {e}"} ]} # Server info try: server_info = r.info() except Exception as e: findings.append({"check": "Server info", "status": "ERROR", "detail": f"Could not retrieve INFO: {e}"}) r.close() return {"info": info, "findings": findings} # Engine and version redis_version = str(server_info.get("redis_version", "unknown")) info["redis_version"] = redis_version info["os"] = server_info.get("os", "unknown") info["uptime_days"] = round(server_info.get("uptime_in_seconds", 0) / 86400, 1) parts = redis_version.split(".") try: version_tuple = tuple(int(p) for p in parts[:2]) if len(parts) >= 2 else (int(parts[0]), 0) except (ValueError, IndexError): version_tuple = (0, 0) if version_tuple >= (7, 0): findings.append({"check": "Engine version", "status": "PASS", "detail": f"Redis {redis_version} can migrate to Valkey 7.2 (direct in-place switch) or Redis OSS 7+ on ElastiCache"}) elif version_tuple >= (6, 0): findings.append({"check": "Engine version", "status": "WARN", "detail": f"Redis {redis_version} standard support ends January 31, 2027. Extended Support charges begin February 1, 2027. Recommend upgrading to Valkey before then."}) elif version_tuple >= (5, 0): findings.append({"check": "Engine version", "status": "FAIL", "detail": f"Redis {redis_version} standard support ended January 31, 2026. Extended Support charges are now active (since Feb 1, 2026). Upgrade to Valkey to eliminate surcharges."}) findings.append({"check": "Extended Support", "status": "FAIL", "detail": f"Redis {redis_version} is enrolled in Extended Support. Year 1-2 premium: 80% surcharge. Year 3 premium: 160% surcharge. Run scripts/price_calculator.py --extended-support to see costs."}) else: findings.append({"check": "Engine version", "status": "FAIL", "detail": f"Redis {redis_version} standard support ended January 31, 2026. Extended Support charges are now active. Upgrade to Valkey to eliminate surcharges."}) findings.append({"check": "Extended Support", "status": "FAIL", "detail": f"Redis {redis_version} is enrolled in Extended Support. Year 1-2 premium: 80% surcharge. Year 3 premium: 160% surcharge. Run scripts/price_calculator.py --extended-support to see costs."}) # Cluster mode cluster_enabled = server_info.get("cluster_enabled", 0) info["cluster_mode"] = bool(cluster_enabled) if cluster_enabled: findings.append({"check": "Cluster mode", "status": "INFO", "detail": "Cluster mode enabled — target must also be cluster mode enabled"}) else: findings.append({"check": "Cluster mode", "status": "INFO", "detail": "Standalone or replica mode — can target cluster mode disabled or enabled"}) # Connected replicas connected_replicas = server_info.get("connected_slaves", 0) info["connected_replicas"] = connected_replicas findings.append({"check": "Replication", "status": "INFO", "detail": f"{connected_replicas} connected replica(s)"}) # Memory used_memory_mb = round(server_info.get("used_memory", 0) / (1024 * 1024), 1) used_memory_gb = round(used_memory_mb / 1024, 2) peak_memory_mb = round(server_info.get("used_memory_peak", 0) / (1024 * 1024), 1) info["used_memory_mb"] = used_memory_mb info["used_memory_gb"] = used_memory_gb info["peak_memory_mb"] = peak_memory_mb findings.append({"check": "Memory usage", "status": "INFO", "detail": f"Current: {used_memory_mb} MB ({used_memory_gb} GB), Peak: {peak_memory_mb} MB"}) # Key count try: db_info = {k: v for k, v in server_info.items() if k.startswith("db")} total_keys = sum(v.get("keys", 0) for v in db_info.values() if isinstance(v, dict)) info["total_keys"] = total_keys info["databases_in_use"] = len(db_info) findings.append({"check": "Key count", "status": "INFO", "detail": f"{total_keys:,} keys across {len(db_info)} database(s)"}) if len(db_info) > 1: findings.append({"check": "Multiple databases", "status": "WARN", "detail": f"{len(db_info)} databases in use — ElastiCache cluster mode uses only db0. Plan key migration."}) except Exception: info["total_keys"] = "unknown" # Persistence rdb_enabled = server_info.get("rdb_last_save_time", 0) > 0 aof_enabled = server_info.get("aof_enabled", 0) == 1 info["rdb_enabled"] = rdb_enabled info["aof_enabled"] = aof_enabled persistence_detail = [] if rdb_enabled: persistence_detail.append("RDB snapshots active") if aof_enabled: persistence_detail.append("AOF enabled") if not persistence_detail: persistence_detail.append("No persistence") findings.append({"check": "Persistence", "status": "INFO", "detail": ", ".join(persistence_detail)}) # Modules try: modules = r.execute_command("MODULE", "LIST") loaded_modules = [] for mod in modules: if isinstance(mod, list): mod_name = mod[1] if len(mod) > 1 else str(mod) elif isinstance(mod, dict): mod_name = mod.get("name", str(mod)) else: mod_name = str(mod) loaded_modules.append(mod_name) info["modules"] = loaded_modules if not loaded_modules: findings.append({"check": "Modules", "status": "PASS", "detail": "No modules loaded"}) else: for mod_name in loaded_modules: compat = MODULE_COMPATIBILITY.get(mod_name) if compat: if compat["elasticache"]: findings.append({"check": f"Module: {compat['name']}", "status": "INFO", "detail": str(compat["note"])}) else: findings.append({"check": f"Module: {compat['name']}", "status": "FAIL", "detail": str(compat["note"])}) else: findings.append({"check": f"Module: {mod_name}", "status": "WARN", "detail": f"Unknown module '{mod_name}' — verify compatibility with ElastiCache"}) except redis.ResponseError: info["modules"] = [] findings.append({"check": "Modules", "status": "INFO", "detail": "MODULE LIST not available (may be restricted or old version)"}) # Commandstats — check for restricted command usage cmdstats = {} try: cmdstats = r.info("commandstats") for cmd_name in ELASTICACHE_RESTRICTED_COMMANDS: stat_key = f"cmdstat_{cmd_name.lower().replace(' ', '|')}" if stat_key in cmdstats: calls = cmdstats[stat_key].get("calls", 0) if calls > 0: findings.append({"check": "Restricted command usage", "status": "WARN", "detail": f"Command {cmd_name} called {calls} times. This command is restricted on ElastiCache. Review usage before migration."}) except redis.ResponseError: findings.append({"check": "Restricted command usage", "status": "INFO", "detail": "INFO COMMANDSTATS not available (may be restricted by ACL or unsupported on this engine version)"}) except (redis.ConnectionError, redis.TimeoutError) as e: findings.append({"check": "Restricted command usage", "status": "INFO", "detail": f"Could not retrieve commandstats: {e}"}) # Lua script detection lua_detected = False try: eval_calls = 0 for key in ("cmdstat_eval", "cmdstat_evalsha"): if key in cmdstats: eval_calls += cmdstats[key].get("calls", 0) if eval_calls > 0: lua_detected = True findings.append({"check": "Lua scripts", "status": "WARN", "detail": f"Lua scripts detected ({eval_calls} eval calls). Test all Lua scripts against the target engine. Scripts using module commands or hardcoded key names may break."}) except (KeyError, TypeError, AttributeError): findings.append({"check": "Lua scripts", "status": "INFO", "detail": "Unable to detect Lua script usage from commandstats"}) info["lua_scripts_detected"] = lua_detected # Large key sampling try: large_keys_found = [] for _ in range(20): key = r.randomkey() if key is None: break try: usage = r.memory_usage(key) if usage is not None and usage > 1048576: size_mb = round(usage / (1024 * 1024), 2) large_keys_found.append((key, size_mb)) except Exception: findings.append({"check": "Large key sampling", "status": "INFO", "detail": "MEMORY USAGE command not available, skipping large key check"}) break for key, size_mb in large_keys_found: key_type = r.type(key) if key else "unknown" findings.append({"check": "Large key sampling", "status": "WARN", "detail": f"Large key detected ({key_type}, {size_mb}MB). Large keys can cause replication timeouts and event loop blocking during migration."}) except Exception: pass # Maxmemory policy # CONFIG GET is restricted on ElastiCache managed endpoints. Detect them by # hostname and skip CONFIG in that case, advising the user to check via the # AWS API (describe-cache-parameters) instead. is_elasticache = host.endswith(".cache.amazonaws.com") if is_elasticache: info["maxmemory_policy"] = "unknown (ElastiCache source)" findings.append({"check": "Eviction policy", "status": "INFO", "detail": "CONFIG command is restricted on ElastiCache. " "Check the eviction policy via: aws elasticache describe-cache-parameters " "--cache-parameter-group-name " "--query \"Parameters[?ParameterName=='maxmemory-policy']\""}) else: try: config = r.config_get("maxmemory-policy") policy = config.get("maxmemory-policy", "unknown") info["maxmemory_policy"] = policy if policy in ("noeviction",): findings.append({"check": "Eviction policy", "status": "WARN", "detail": f"Policy is '{policy}' -- writes will fail when memory is full. Consider volatile-lru or allkeys-lru."}) else: findings.append({"check": "Eviction policy", "status": "INFO", "detail": f"Policy: {policy}"}) except redis.ResponseError: info["maxmemory_policy"] = "unknown" # Data tiering advisory findings.append({"check": "Data tiering advisory", "status": "INFO", "detail": "If targeting r6gd node types (data tiering), note: online migration is not supported for r6gd clusters. " "Use backup/restore instead. Data tiering only supports volatile-lru, allkeys-lru, volatile-lfu, allkeys-lfu, and noeviction eviction policies."}) # Sizing recommendation findings.append({"check": "Sizing recommendation", "status": "INFO", "detail": f"Source dataset: {used_memory_gb} GB. Use scripts/price_calculator.py to estimate serverless and node-based costs for your workload."}) try: r.close() except Exception: pass return {"info": info, "findings": findings} def format_report(result): """Format preflight results as a human-readable report.""" if "error" in result: info = result.get("info", {}) lines = [f"ERROR: {result['error']}"] if info: lines.append(f" Host: {info.get('host', '?')}:{info.get('port', '?')}") return "\n".join(lines) info = result["info"] findings = result["findings"] lines = [] lines.append("=" * 72) lines.append("ElastiCache Migration Preflight Report") lines.append("=" * 72) lines.append("") lines.append(f" Source: {info['host']}:{info['port']}") lines.append(f" Redis version: {info.get('redis_version', 'unknown')}") lines.append(f" OS: {info.get('os', 'unknown')}") lines.append(f" Uptime: {info.get('uptime_days', '?')} days") lines.append(f" Cluster mode: {'enabled' if info.get('cluster_mode') else 'disabled'}") lines.append(f" Memory: {info.get('used_memory_mb', '?')} MB ({info.get('used_memory_gb', '?')} GB)") lines.append(f" Keys: {info.get('total_keys', '?')}") lines.append(f" Replicas: {info.get('connected_replicas', '?')}") lines.append(f" Modules: {', '.join(info.get('modules', [])) or 'none'}") lines.append("") fail_count = sum(1 for f in findings if f["status"] == "FAIL") warn_count = sum(1 for f in findings if f["status"] == "WARN") pass_count = sum(1 for f in findings if f["status"] == "PASS") if fail_count == 0: lines.append(f" Verdict: READY TO MIGRATE ({pass_count} passed, {warn_count} warnings)") else: lines.append(f" Verdict: BLOCKERS FOUND ({fail_count} failures, {warn_count} warnings)") lines.append("") lines.append("-" * 72) status_order = {"FAIL": 0, "WARN": 1, "ERROR": 2, "PASS": 3, "INFO": 4} sorted_findings = sorted(findings, key=lambda f: status_order.get(f["status"], 5)) for f in sorted_findings: icon = {"PASS": "[OK] ", "FAIL": "[FAIL]", "WARN": "[WARN]", "INFO": "[INFO]", "ERROR": "[ERR] "} lines.append(f" {icon.get(f['status'], '[?] ')} {f['check']}") lines.append(f" {f['detail']}") lines.append("") if fail_count > 0: lines.append("=" * 72) lines.append("MIGRATION BLOCKERS") lines.append("=" * 72) lines.append("") for f in sorted_findings: if f["status"] == "FAIL": lines.append(f" - {f['check']}: {f['detail']}") lines.append("") lines.append(" Resolve blockers before starting migration.") lines.append(" See references/migration/instructions.md for guidance.") lines.append("") lines.append(" Next steps:") lines.append(" 1. Resolve any blockers above") lines.append(" 2. Run: python scripts/price_calculator.py to estimate target cost") lines.append(" 3. Use the AWS CLI test-migration command to validate connectivity to the target cluster") lines.append(" 4. See references/migration/instructions.md for the full migration workflow") lines.append("") return "\n".join(lines) if __name__ == "__main__": parser = argparse.ArgumentParser(description="ElastiCache Migration Preflight Check") parser.add_argument("--host", required=True, help="Source Redis/Valkey hostname") parser.add_argument("--port", type=int, default=6379, help="Port (default: 6379)") parser.add_argument("--password", default=None, help="AUTH password") parser.add_argument("--username", default=None, help="Username (for ACL-based auth)") parser.add_argument("--tls", action="store_true", help="Use TLS") parser.add_argument("--json", action="store_true", help="Output as JSON") args = parser.parse_args() result = run_preflight(args.host, args.port, args.password, args.tls, args.username) if args.json: print(json.dumps(result, indent=2, default=str)) else: print(format_report(result)) if "error" in result: sys.exit(1) fail_count = sum(1 for f in result.get("findings", []) if f["status"] == "FAIL") sys.exit(1 if fail_count > 0 else 0)