ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, name):\n return False, \"Cache name may only contain lowercase alphanumeric characters and hyphens\"\n if '--' in name:\n return False, \"Cache name must not contain consecutive hyphens\"\n if name.endswith('-'):\n return False, \"Cache name must not end with a hyphen\"\n return True, None\n\n\ndef validate_replication_group_id(rg_id: str) -> tuple[bool, Optional[str]]:\n \"\"\"Validate a replication group ID.\n\n Same rules as cache name: 1-40 chars, alphanumeric + hyphens, starts with\n a letter, no consecutive hyphens, no trailing hyphen.\n \"\"\"\n if not rg_id:\n return False, \"Replication group ID must not be empty\"\n if len(rg_id) > 40:\n return False, f\"Replication group ID must be 1-40 characters (got {len(rg_id)})\"\n if not re.match(r'^[a-zA-Z]', rg_id):\n return False, \"Replication group ID must start with a letter\"\n if not re.match(r'^[a-zA-Z0-9-]+

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, rg_id):\n return False, \"Replication group ID may only contain alphanumeric characters and hyphens\"\n if '--' in rg_id:\n return False, \"Replication group ID must not contain consecutive hyphens\"\n if rg_id.endswith('-'):\n return False, \"Replication group ID must not end with a hyphen\"\n return True, None\n\n\ndef validate_subnet_id(subnet_id: str) -> tuple[bool, Optional[str]]:\n \"\"\"Validate an AWS subnet ID (subnet-\u003c8-17 hex chars>).\"\"\"\n if not subnet_id:\n return False, \"Subnet ID must not be empty\"\n if not re.match(r'^subnet-[0-9a-f]{8,17}

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, subnet_id):\n return False, \"Subnet ID must match pattern subnet-\u003c8-17 hex characters> (e.g. subnet-0abc1234def56789a)\"\n return True, None\n\n\ndef validate_security_group_id(sg_id: str) -> tuple[bool, Optional[str]]:\n \"\"\"Validate an AWS security group ID (sg-\u003c8-17 hex chars>).\"\"\"\n if not sg_id:\n return False, \"Security group ID must not be empty\"\n if not re.match(r'^sg-[0-9a-f]{8,17}

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, sg_id):\n return False, \"Security group ID must match pattern sg-\u003c8-17 hex characters> (e.g. sg-0abc1234def56789a)\"\n return True, None\n\n\ndef validate_region(region: str) -> tuple[bool, Optional[str]]:\n \"\"\"Validate an AWS region code (e.g. us-east-1, eu-west-2, ap-southeast-1).\"\"\"\n if not region:\n return False, \"Region must not be empty\"\n if not re.match(r'^[a-z]{2}-[a-z]+-\\d+

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, region):\n return False, \"Region must match AWS region pattern (e.g. us-east-1, eu-west-2)\"\n return True, None\n\n\ndef validate_engine_version(version: str) -> tuple[bool, Optional[str]]:\n r\"\"\"Validate an engine version string.\n\n Accepted formats:\n - Major only: \"7\" (used by serverless MajorEngineVersion)\n - Major.minor: \"7.1\", \"8.2\"\n - Major.minor.patch: \"7.0.4\", \"7.1.0\"\n \"\"\"\n if not version:\n return False, \"Engine version must not be empty\"\n if not re.match(r'^\\d+(\\.\\d+){0,2}

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, version):\n return False, \"Engine version must match pattern \u003cmajor>[.\u003cminor>[.\u003cpatch>]] (e.g. 7, 7.1, 7.0.4)\"\n return True, None\n\n\ndef validate_vpc_id(vpc_id: str) -> tuple[bool, Optional[str]]:\n \"\"\"Validate an AWS VPC ID (vpc-\u003c8-17 hex chars>).\"\"\"\n if not vpc_id:\n return False, \"VPC ID must not be empty\"\n if not re.match(r'^vpc-[0-9a-f]{8,17}

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, vpc_id):\n return False, \"VPC ID must match pattern vpc-\u003c8-17 hex characters> (e.g. vpc-0abc1234def56789a)\"\n return True, None\n\n\ndef validate_kms_key_id(key_id: str) -> tuple[bool, Optional[str]]:\n \"\"\"Validate a KMS key ID.\n\n Accepts either:\n - A UUID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx\n - A KMS key ARN: arn:aws:kms:\u003cregion>:\u003caccount>:key/\u003ckey-id>\n - A KMS alias ARN: arn:aws:kms:\u003cregion>:\u003caccount>:alias/\u003calias-name>\n \"\"\"\n if not key_id:\n return False, \"KMS key ID must not be empty\"\n\n uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

\n arn_pattern = r'^arn:aws:kms:[a-z0-9-]+:\\d{12}:(key|alias)/[a-zA-Z0-9/_-]+

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

\n\n if re.match(uuid_pattern, key_id):\n return True, None\n if re.match(arn_pattern, key_id):\n return True, None\n\n return False, (\n \"KMS key ID must be a UUID (e.g. 12345678-1234-1234-1234-123456789abc) \"\n \"or a KMS ARN (e.g. arn:aws:kms:us-east-1:123456789012:key/\u003ckey-id>)\"\n )\n\n\ndef validate_snapshot_name(name: str, serverless: bool = False) -> tuple[bool, Optional[str]]:\n \"\"\"Validate a snapshot name.\n\n Rules:\n - 1 to 255 characters for serverless snapshots, 1 to 40 for node-based\n - Alphanumeric characters and hyphens only\n - Must start with a letter\n - No consecutive hyphens\n - Must not end with a hyphen\n \"\"\"\n if not name:\n return False, \"Snapshot name must not be empty\"\n max_len = 255 if serverless else 40\n if len(name) > max_len:\n label = \"serverless\" if serverless else \"node-based\"\n return False, f\"Snapshot name must be 1-{max_len} characters for {label} (got {len(name)})\"\n if not re.match(r'^[a-zA-Z]', name):\n return False, \"Snapshot name must start with a letter\"\n if not re.match(r'^[a-zA-Z0-9-]+

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…

, name):\n return False, \"Snapshot name may only contain alphanumeric characters and hyphens\"\n if '--' in name:\n return False, \"Snapshot name must not contain consecutive hyphens\"\n if name.endswith('-'):\n return False, \"Snapshot name must not end with a hyphen\"\n return True, None\n\n\n# ---------------------------------------------------------------------------\n# CLI sanitization safety net\n# ---------------------------------------------------------------------------\n\n# Shell metacharacters that could enable command injection\n_SHELL_METACHARACTERS = re.compile(r'[;&|`$(){}!\u003c>\\'\\\"\\\\\\n\\r\\x00]')\n\n\ndef sanitize_for_cli(value: str) -> str:\n \"\"\"Strip shell metacharacters from a value as a defense-in-depth safety net.\n\n This is NOT a substitute for proper validation -- always validate inputs\n first using the validate_* functions. This function provides an additional\n layer of protection by removing characters that could be interpreted by\n a shell if a value is accidentally interpolated into a command string.\n\n Returns the sanitized string with all shell metacharacters removed.\n \"\"\"\n return _SHELL_METACHARACTERS.sub('', value)\n\n\n# ---------------------------------------------------------------------------\n# Bulk validation\n# ---------------------------------------------------------------------------\n\n# Map of recognized parameter keys to their validator functions\n_VALIDATORS = {\n \"cache_name\": validate_cache_name,\n \"replication_group_id\": validate_replication_group_id,\n \"subnet_id\": validate_subnet_id,\n \"security_group_id\": validate_security_group_id,\n \"region\": validate_region,\n \"engine_version\": validate_engine_version,\n \"vpc_id\": validate_vpc_id,\n \"kms_key_id\": validate_kms_key_id,\n \"snapshot_name\": validate_snapshot_name,\n}\n\n\ndef validate_all(params: dict) -> list[str]:\n \"\"\"Validate all recognized keys in a parameter dict.\n\n Takes a dict like {\"cache_name\": \"my-cache\", \"region\": \"us-east-1\"} and\n validates every key that has a known validator. Keys not recognized are\n silently skipped (they may be application-specific parameters).\n\n Returns a list of error messages. An empty list means all recognized\n parameters passed validation.\n \"\"\"\n errors = []\n for key, value in params.items():\n validator = _VALIDATORS.get(key)\n if validator is None:\n continue\n if not isinstance(value, str):\n errors.append(f\"{key}: expected a string, got {type(value).__name__}\")\n continue\n is_valid, error_msg = validator(value) # type: ignore[operator]\n if not is_valid:\n errors.append(f\"{key}: {error_msg}\")\n return errors\n\n\n# ---------------------------------------------------------------------------\n# CLI entrypoint\n# ---------------------------------------------------------------------------\n\ndef main():\n \"\"\"Validate key=value pairs passed on the command line.\"\"\"\n if len(sys.argv) \u003c 2:\n print(\"Usage: python input_validator.py key=value [key=value ...]\")\n print(\"Example: python input_validator.py cache_name=my-cache region=us-east-1\")\n print()\n print(\"Recognized keys:\", \", \".join(sorted(_VALIDATORS.keys())))\n sys.exit(0)\n\n params = {}\n for arg in sys.argv[1:]:\n if '=' not in arg:\n print(f\"Error: argument '{arg}' is not in key=value format\")\n sys.exit(2)\n key, value = arg.split('=', 1)\n params[key] = value\n\n errors = validate_all(params)\n\n if errors:\n print(\"Validation FAILED:\")\n for err in errors:\n print(f\" - {err}\")\n sys.exit(1)\n else:\n print(\"All inputs valid.\")\n sys.exit(0)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":10876,"content_sha256":"26c38390ebf7e60b2559531c41340d52630c3e872402b66dc780c1592b8d0ad0"},{"filename":"scripts/migration_preflight.py","content":"#!/usr/bin/env python3\n\"\"\"\nElastiCache Migration Preflight Check\n\nConnects to a source Redis/Valkey instance and evaluates migration readiness:\n- Engine version compatibility\n- Module detection (RedisJSON, RediSearch, RedisTimeSeries, etc.)\n- Key count and memory usage\n- Persistence configuration\n- Cluster mode detection\n- Unsupported command sampling\n\nUsage:\n # Check a source Redis instance (plain)\n python migration_preflight.py --host redis.example.com --port 6379\n\n # With TLS\n python migration_preflight.py --host redis.example.com --port 6379 --tls\n\n # With auth\n python migration_preflight.py --host redis.example.com --password mypassword\n\n # JSON output\n python migration_preflight.py --host redis.example.com --json\n\n # Via tunnel (local port forward)\n python migration_preflight.py --host localhost --port 6379\n\nRequires: valkey (pip install valkey)\n\"\"\"\n\nimport argparse\nimport json\nimport sys\n\n# ---------------------------------------------------------------------------\n# Dependency check -- fail early with an actionable message instead of a\n# raw ImportError traceback.\n# ---------------------------------------------------------------------------\ntry:\n import valkey as redis\nexcept ImportError:\n print(\n \"Error: the 'valkey' package is required for this script.\\n\"\n \"Install with:\\n\"\n \" pip install valkey\"\n )\n sys.exit(1)\n\n\n# Commands not available on ElastiCache (restricted or modified)\nELASTICACHE_RESTRICTED_COMMANDS = {\n \"BGSAVE\", \"BGREWRITEAOF\", \"CONFIG\", \"DEBUG\", \"MIGRATE\",\n \"SAVE\", \"SHUTDOWN\", \"SLAVEOF\", \"REPLICAOF\", \"SYNC\",\n}\n\n# Known Redis modules and their ElastiCache availability\nMODULE_COMPATIBILITY = {\n \"search\": {\"name\": \"RediSearch\", \"elasticache\": False, \"note\": \"Not available on ElastiCache. Valkey 8.2 provides native vector search (FT.* commands) on node-based clusters.\"},\n \"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.\"},\n \"timeseries\": {\"name\": \"RedisTimeSeries\", \"elasticache\": False, \"note\": \"Not available on ElastiCache. Consider CloudWatch or Timestream.\"},\n \"bf\": {\"name\": \"RedisBloom\", \"elasticache\": False, \"note\": \"No (module) / Yes (native BF.* commands in Valkey 8.1+).\"},\n \"graph\": {\"name\": \"RedisGraph\", \"elasticache\": False, \"note\": \"Deprecated. Consider Neptune or OpenSearch.\"},\n \"ai\": {\"name\": \"RedisAI\", \"elasticache\": False, \"note\": \"Not available on ElastiCache. Use SageMaker for ML inference.\"},\n}\n\n\ndef run_preflight(host, port, password=None, use_tls=False, username=None):\n \"\"\"Run all preflight checks against a source Redis/Valkey instance.\"\"\"\n findings = []\n info = {\"host\": host, \"port\": port}\n\n try:\n r = redis.Redis(\n host=host, port=port,\n password=password, username=username,\n ssl=use_tls,\n decode_responses=True,\n socket_timeout=10,\n socket_connect_timeout=10,\n )\n r.ping()\n findings.append({\"check\": \"Connectivity\", \"status\": \"PASS\", \"detail\": f\"Connected to {host}:{port}\"})\n except (redis.ConnectionError, redis.TimeoutError) as e:\n return {\"error\": f\"Cannot connect to {host}:{port}: {e}\", \"info\": info, \"findings\": [\n {\"check\": \"Connectivity\", \"status\": \"FAIL\", \"detail\": str(e)}\n ]}\n except redis.AuthenticationError as e:\n return {\"error\": f\"Authentication failed: {e}\", \"info\": info, \"findings\": [\n {\"check\": \"Connectivity\", \"status\": \"FAIL\", \"detail\": f\"Auth failed: {e}\"}\n ]}\n\n # Server info\n try:\n server_info = r.info()\n except Exception as e:\n findings.append({\"check\": \"Server info\", \"status\": \"ERROR\", \"detail\": f\"Could not retrieve INFO: {e}\"})\n r.close()\n return {\"info\": info, \"findings\": findings}\n\n # Engine and version\n redis_version = str(server_info.get(\"redis_version\", \"unknown\"))\n info[\"redis_version\"] = redis_version\n info[\"os\"] = server_info.get(\"os\", \"unknown\")\n info[\"uptime_days\"] = round(server_info.get(\"uptime_in_seconds\", 0) / 86400, 1)\n\n parts = redis_version.split(\".\")\n try:\n version_tuple = tuple(int(p) for p in parts[:2]) if len(parts) >= 2 else (int(parts[0]), 0)\n except (ValueError, IndexError):\n version_tuple = (0, 0)\n\n if version_tuple >= (7, 0):\n findings.append({\"check\": \"Engine version\", \"status\": \"PASS\",\n \"detail\": f\"Redis {redis_version} can migrate to Valkey 7.2 (direct in-place switch) or Redis OSS 7+ on ElastiCache\"})\n elif version_tuple >= (6, 0):\n findings.append({\"check\": \"Engine version\", \"status\": \"WARN\",\n \"detail\": f\"Redis {redis_version} standard support ends January 31, 2027. Extended Support charges begin February 1, 2027. Recommend upgrading to Valkey before then.\"})\n elif version_tuple >= (5, 0):\n findings.append({\"check\": \"Engine version\", \"status\": \"FAIL\",\n \"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.\"})\n findings.append({\"check\": \"Extended Support\", \"status\": \"FAIL\",\n \"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.\"})\n else:\n findings.append({\"check\": \"Engine version\", \"status\": \"FAIL\",\n \"detail\": f\"Redis {redis_version} standard support ended January 31, 2026. Extended Support charges are now active. Upgrade to Valkey to eliminate surcharges.\"})\n findings.append({\"check\": \"Extended Support\", \"status\": \"FAIL\",\n \"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.\"})\n\n # Cluster mode\n cluster_enabled = server_info.get(\"cluster_enabled\", 0)\n info[\"cluster_mode\"] = bool(cluster_enabled)\n if cluster_enabled:\n findings.append({\"check\": \"Cluster mode\", \"status\": \"INFO\",\n \"detail\": \"Cluster mode enabled — target must also be cluster mode enabled\"})\n else:\n findings.append({\"check\": \"Cluster mode\", \"status\": \"INFO\",\n \"detail\": \"Standalone or replica mode — can target cluster mode disabled or enabled\"})\n\n # Connected replicas\n connected_replicas = server_info.get(\"connected_slaves\", 0)\n info[\"connected_replicas\"] = connected_replicas\n findings.append({\"check\": \"Replication\", \"status\": \"INFO\",\n \"detail\": f\"{connected_replicas} connected replica(s)\"})\n\n # Memory\n used_memory_mb = round(server_info.get(\"used_memory\", 0) / (1024 * 1024), 1)\n used_memory_gb = round(used_memory_mb / 1024, 2)\n peak_memory_mb = round(server_info.get(\"used_memory_peak\", 0) / (1024 * 1024), 1)\n info[\"used_memory_mb\"] = used_memory_mb\n info[\"used_memory_gb\"] = used_memory_gb\n info[\"peak_memory_mb\"] = peak_memory_mb\n findings.append({\"check\": \"Memory usage\", \"status\": \"INFO\",\n \"detail\": f\"Current: {used_memory_mb} MB ({used_memory_gb} GB), Peak: {peak_memory_mb} MB\"})\n\n # Key count\n try:\n db_info = {k: v for k, v in server_info.items() if k.startswith(\"db\")}\n total_keys = sum(v.get(\"keys\", 0) for v in db_info.values() if isinstance(v, dict))\n info[\"total_keys\"] = total_keys\n info[\"databases_in_use\"] = len(db_info)\n findings.append({\"check\": \"Key count\", \"status\": \"INFO\",\n \"detail\": f\"{total_keys:,} keys across {len(db_info)} database(s)\"})\n\n if len(db_info) > 1:\n findings.append({\"check\": \"Multiple databases\", \"status\": \"WARN\",\n \"detail\": f\"{len(db_info)} databases in use — ElastiCache cluster mode uses only db0. Plan key migration.\"})\n except Exception:\n info[\"total_keys\"] = \"unknown\"\n\n # Persistence\n rdb_enabled = server_info.get(\"rdb_last_save_time\", 0) > 0\n aof_enabled = server_info.get(\"aof_enabled\", 0) == 1\n info[\"rdb_enabled\"] = rdb_enabled\n info[\"aof_enabled\"] = aof_enabled\n persistence_detail = []\n if rdb_enabled:\n persistence_detail.append(\"RDB snapshots active\")\n if aof_enabled:\n persistence_detail.append(\"AOF enabled\")\n if not persistence_detail:\n persistence_detail.append(\"No persistence\")\n findings.append({\"check\": \"Persistence\", \"status\": \"INFO\",\n \"detail\": \", \".join(persistence_detail)})\n\n # Modules\n try:\n modules = r.execute_command(\"MODULE\", \"LIST\")\n loaded_modules = []\n for mod in modules:\n if isinstance(mod, list):\n mod_name = mod[1] if len(mod) > 1 else str(mod)\n elif isinstance(mod, dict):\n mod_name = mod.get(\"name\", str(mod))\n else:\n mod_name = str(mod)\n loaded_modules.append(mod_name)\n\n info[\"modules\"] = loaded_modules\n\n if not loaded_modules:\n findings.append({\"check\": \"Modules\", \"status\": \"PASS\", \"detail\": \"No modules loaded\"})\n else:\n for mod_name in loaded_modules:\n compat = MODULE_COMPATIBILITY.get(mod_name)\n if compat:\n if compat[\"elasticache\"]:\n findings.append({\"check\": f\"Module: {compat['name']}\", \"status\": \"INFO\",\n \"detail\": str(compat[\"note\"])})\n else:\n findings.append({\"check\": f\"Module: {compat['name']}\", \"status\": \"FAIL\",\n \"detail\": str(compat[\"note\"])})\n else:\n findings.append({\"check\": f\"Module: {mod_name}\", \"status\": \"WARN\",\n \"detail\": f\"Unknown module '{mod_name}' — verify compatibility with ElastiCache\"})\n except redis.ResponseError:\n info[\"modules\"] = []\n findings.append({\"check\": \"Modules\", \"status\": \"INFO\",\n \"detail\": \"MODULE LIST not available (may be restricted or old version)\"})\n\n # Commandstats — check for restricted command usage\n cmdstats = {}\n try:\n cmdstats = r.info(\"commandstats\")\n for cmd_name in ELASTICACHE_RESTRICTED_COMMANDS:\n stat_key = f\"cmdstat_{cmd_name.lower().replace(' ', '|')}\"\n if stat_key in cmdstats:\n calls = cmdstats[stat_key].get(\"calls\", 0)\n if calls > 0:\n findings.append({\"check\": \"Restricted command usage\", \"status\": \"WARN\",\n \"detail\": f\"Command {cmd_name} called {calls} times. This command is restricted on ElastiCache. Review usage before migration.\"})\n except redis.ResponseError:\n findings.append({\"check\": \"Restricted command usage\", \"status\": \"INFO\",\n \"detail\": \"INFO COMMANDSTATS not available (may be restricted by ACL or unsupported on this engine version)\"})\n except (redis.ConnectionError, redis.TimeoutError) as e:\n findings.append({\"check\": \"Restricted command usage\", \"status\": \"INFO\",\n \"detail\": f\"Could not retrieve commandstats: {e}\"})\n\n # Lua script detection\n lua_detected = False\n try:\n eval_calls = 0\n for key in (\"cmdstat_eval\", \"cmdstat_evalsha\"):\n if key in cmdstats:\n eval_calls += cmdstats[key].get(\"calls\", 0)\n if eval_calls > 0:\n lua_detected = True\n findings.append({\"check\": \"Lua scripts\", \"status\": \"WARN\",\n \"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.\"})\n except (KeyError, TypeError, AttributeError):\n findings.append({\"check\": \"Lua scripts\", \"status\": \"INFO\",\n \"detail\": \"Unable to detect Lua script usage from commandstats\"})\n info[\"lua_scripts_detected\"] = lua_detected\n\n # Large key sampling\n try:\n large_keys_found = []\n for _ in range(20):\n key = r.randomkey()\n if key is None:\n break\n try:\n usage = r.memory_usage(key)\n if usage is not None and usage > 1048576:\n size_mb = round(usage / (1024 * 1024), 2)\n large_keys_found.append((key, size_mb))\n except Exception:\n findings.append({\"check\": \"Large key sampling\", \"status\": \"INFO\",\n \"detail\": \"MEMORY USAGE command not available, skipping large key check\"})\n break\n for key, size_mb in large_keys_found:\n key_type = r.type(key) if key else \"unknown\"\n findings.append({\"check\": \"Large key sampling\", \"status\": \"WARN\",\n \"detail\": f\"Large key detected ({key_type}, {size_mb}MB). Large keys can cause replication timeouts and event loop blocking during migration.\"})\n except Exception:\n pass\n\n # Maxmemory policy\n # CONFIG GET is restricted on ElastiCache managed endpoints. Detect them by\n # hostname and skip CONFIG in that case, advising the user to check via the\n # AWS API (describe-cache-parameters) instead.\n is_elasticache = host.endswith(\".cache.amazonaws.com\")\n if is_elasticache:\n info[\"maxmemory_policy\"] = \"unknown (ElastiCache source)\"\n findings.append({\"check\": \"Eviction policy\", \"status\": \"INFO\",\n \"detail\": \"CONFIG command is restricted on ElastiCache. \"\n \"Check the eviction policy via: aws elasticache describe-cache-parameters \"\n \"--cache-parameter-group-name \u003cparam-group> \"\n \"--query \\\"Parameters[?ParameterName=='maxmemory-policy']\\\"\"})\n else:\n try:\n config = r.config_get(\"maxmemory-policy\")\n policy = config.get(\"maxmemory-policy\", \"unknown\")\n info[\"maxmemory_policy\"] = policy\n if policy in (\"noeviction\",):\n findings.append({\"check\": \"Eviction policy\", \"status\": \"WARN\",\n \"detail\": f\"Policy is '{policy}' -- writes will fail when memory is full. Consider volatile-lru or allkeys-lru.\"})\n else:\n findings.append({\"check\": \"Eviction policy\", \"status\": \"INFO\",\n \"detail\": f\"Policy: {policy}\"})\n except redis.ResponseError:\n info[\"maxmemory_policy\"] = \"unknown\"\n\n # Data tiering advisory\n findings.append({\"check\": \"Data tiering advisory\", \"status\": \"INFO\",\n \"detail\": \"If targeting r6gd node types (data tiering), note: online migration is not supported for r6gd clusters. \"\n \"Use backup/restore instead. Data tiering only supports volatile-lru, allkeys-lru, volatile-lfu, allkeys-lfu, and noeviction eviction policies.\"})\n\n # Sizing recommendation\n findings.append({\"check\": \"Sizing recommendation\", \"status\": \"INFO\",\n \"detail\": f\"Source dataset: {used_memory_gb} GB. Use scripts/price_calculator.py to estimate serverless and node-based costs for your workload.\"})\n\n try:\n r.close()\n except Exception:\n pass\n return {\"info\": info, \"findings\": findings}\n\n\ndef format_report(result):\n \"\"\"Format preflight results as a human-readable report.\"\"\"\n if \"error\" in result:\n info = result.get(\"info\", {})\n lines = [f\"ERROR: {result['error']}\"]\n if info:\n lines.append(f\" Host: {info.get('host', '?')}:{info.get('port', '?')}\")\n return \"\\n\".join(lines)\n\n info = result[\"info\"]\n findings = result[\"findings\"]\n\n lines = []\n lines.append(\"=\" * 72)\n lines.append(\"ElastiCache Migration Preflight Report\")\n lines.append(\"=\" * 72)\n lines.append(\"\")\n lines.append(f\" Source: {info['host']}:{info['port']}\")\n lines.append(f\" Redis version: {info.get('redis_version', 'unknown')}\")\n lines.append(f\" OS: {info.get('os', 'unknown')}\")\n lines.append(f\" Uptime: {info.get('uptime_days', '?')} days\")\n lines.append(f\" Cluster mode: {'enabled' if info.get('cluster_mode') else 'disabled'}\")\n lines.append(f\" Memory: {info.get('used_memory_mb', '?')} MB ({info.get('used_memory_gb', '?')} GB)\")\n lines.append(f\" Keys: {info.get('total_keys', '?')}\")\n lines.append(f\" Replicas: {info.get('connected_replicas', '?')}\")\n lines.append(f\" Modules: {', '.join(info.get('modules', [])) or 'none'}\")\n lines.append(\"\")\n\n fail_count = sum(1 for f in findings if f[\"status\"] == \"FAIL\")\n warn_count = sum(1 for f in findings if f[\"status\"] == \"WARN\")\n pass_count = sum(1 for f in findings if f[\"status\"] == \"PASS\")\n\n if fail_count == 0:\n lines.append(f\" Verdict: READY TO MIGRATE ({pass_count} passed, {warn_count} warnings)\")\n else:\n lines.append(f\" Verdict: BLOCKERS FOUND ({fail_count} failures, {warn_count} warnings)\")\n lines.append(\"\")\n lines.append(\"-\" * 72)\n\n status_order = {\"FAIL\": 0, \"WARN\": 1, \"ERROR\": 2, \"PASS\": 3, \"INFO\": 4}\n sorted_findings = sorted(findings, key=lambda f: status_order.get(f[\"status\"], 5))\n\n for f in sorted_findings:\n icon = {\"PASS\": \"[OK] \", \"FAIL\": \"[FAIL]\", \"WARN\": \"[WARN]\", \"INFO\": \"[INFO]\", \"ERROR\": \"[ERR] \"}\n lines.append(f\" {icon.get(f['status'], '[?] ')} {f['check']}\")\n lines.append(f\" {f['detail']}\")\n lines.append(\"\")\n\n if fail_count > 0:\n lines.append(\"=\" * 72)\n lines.append(\"MIGRATION BLOCKERS\")\n lines.append(\"=\" * 72)\n lines.append(\"\")\n for f in sorted_findings:\n if f[\"status\"] == \"FAIL\":\n lines.append(f\" - {f['check']}: {f['detail']}\")\n lines.append(\"\")\n lines.append(\" Resolve blockers before starting migration.\")\n lines.append(\" See references/migration/instructions.md for guidance.\")\n\n lines.append(\"\")\n lines.append(\" Next steps:\")\n lines.append(\" 1. Resolve any blockers above\")\n lines.append(\" 2. Run: python scripts/price_calculator.py to estimate target cost\")\n lines.append(\" 3. Use the AWS CLI test-migration command to validate connectivity to the target cluster\")\n lines.append(\" 4. See references/migration/instructions.md for the full migration workflow\")\n lines.append(\"\")\n\n return \"\\n\".join(lines)\n\n\nif __name__ == \"__main__\":\n parser = argparse.ArgumentParser(description=\"ElastiCache Migration Preflight Check\")\n parser.add_argument(\"--host\", required=True, help=\"Source Redis/Valkey hostname\")\n parser.add_argument(\"--port\", type=int, default=6379, help=\"Port (default: 6379)\")\n parser.add_argument(\"--password\", default=None, help=\"AUTH password\")\n parser.add_argument(\"--username\", default=None, help=\"Username (for ACL-based auth)\")\n parser.add_argument(\"--tls\", action=\"store_true\", help=\"Use TLS\")\n parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n args = parser.parse_args()\n\n result = run_preflight(args.host, args.port, args.password, args.tls, args.username)\n\n if args.json:\n print(json.dumps(result, indent=2, default=str))\n else:\n print(format_report(result))\n\n if \"error\" in result:\n sys.exit(1)\n fail_count = sum(1 for f in result.get(\"findings\", []) if f[\"status\"] == \"FAIL\")\n sys.exit(1 if fail_count > 0 else 0)\n","content_type":"text/x-python; charset=utf-8","language":"python","size":19948,"content_sha256":"9da42c9ba7e8e9e2b0af7d3b0ed2fb1aedaf3cc6311de68d23e9b44f0bb90338"},{"filename":"scripts/price_calculator.py","content":"#!/usr/bin/env python3\n\"\"\"\nElastiCache Price Calculator\n\nFetches live pricing from AWS and estimates costs for specific configurations.\n\nSupports:\n- Serverless (Valkey, Redis OSS, Memcached)\n- Node-based on-demand and reserved instances\n- Multi-node clusters with replicas\n- Extended Support surcharge for EOL Redis versions\n\nUsage:\n # Serverless estimate\n python3 price_calculator.py --engine valkey --mode serverless --data-gb 5 --ecpu-millions 200\n\n # Node-based with replicas\n python3 price_calculator.py --engine valkey --mode node --node-type cache.r7g.large --nodes 3\n\n # Node-based with specific RI term\n python3 price_calculator.py --mode node --node-type cache.r7g.large --nodes 2 --ri-term 1yr_no_upfront\n\n # Show all reserved options for a node type\n python3 price_calculator.py --mode node --node-type cache.r7g.large --nodes 2 --show-ri-options\n\n # Specific region\n python3 price_calculator.py --engine valkey --region eu-west-1 --mode serverless --data-gb 10\n\n # Extended Support surcharge for EOL Redis versions\n python3 price_calculator.py --engine redis --extended-support --node-type cache.r7g.large --nodes 6\n\n # Interactive mode\n python3 price_calculator.py --interactive\n\"\"\"\n\nimport argparse\nimport json\nimport os\nimport sys\n\nsys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))\nfrom pricing import PricingLoader\n\n# Detect region from environment with us-east-1 as fallback\nREGION = os.environ.get(\"AWS_REGION\") or os.environ.get(\"AWS_DEFAULT_REGION\") or \"us-east-1\"\n\nHOURS_PER_MONTH = 730\n\nPRICING_DISCLAIMER = (\n \" Estimates only. Verify current pricing at https://aws.amazon.com/elasticache/pricing/\"\n)\n\n\ndef estimate_serverless(loader, engine, region, data_gb, ecpu_millions_per_month, avg_payload_kb=1.0):\n storage_rate = loader.get_serverless_storage_rate(region, engine)\n ecpu_rate = loader.get_serverless_ecpu_rate(region, engine)\n min_gb = 0.1 if engine == \"valkey\" else 1.0\n billing_gb = max(data_gb, min_gb)\n storage = billing_gb * storage_rate * HOURS_PER_MONTH\n scaled_ecpus = ecpu_millions_per_month * max(1.0, avg_payload_kb)\n compute = scaled_ecpus * 1_000_000 * ecpu_rate\n result = {\n \"model\": \"serverless\",\n \"engine\": engine,\n \"region\": region,\n \"data_gb\": data_gb,\n \"ecpu_millions\": ecpu_millions_per_month,\n \"storage_monthly\": round(storage, 2),\n \"compute_monthly\": round(compute, 2),\n \"total_monthly\": round(storage + compute, 2),\n }\n if avg_payload_kb != 1.0:\n result[\"avg_payload_kb\"] = avg_payload_kb\n result[\"ecpu_millions_scaled\"] = round(scaled_ecpus, 1)\n return result\n\n\ndef estimate_node_based(loader, engine, region, node_type, node_count, ri_term=None):\n try:\n hourly = loader.get_node_hourly_rate(region, node_type, engine)\n except ValueError as e:\n return {\"error\": str(e)}\n\n monthly_per_node = hourly * HOURS_PER_MONTH\n total = monthly_per_node * node_count\n mem = loader.get_node_memory_gb(node_type)\n\n result = {\n \"model\": \"node-based (on-demand)\",\n \"engine\": engine,\n \"region\": region,\n \"node_type\": node_type,\n \"node_count\": node_count,\n \"memory_per_node_gb\": mem if mem > 0 else \"unknown\",\n \"total_memory_gb\": round(mem * node_count, 1) if mem > 0 else \"unknown\",\n \"per_node_monthly\": round(monthly_per_node, 2),\n \"total_monthly\": round(total, 2),\n }\n\n # Try real reserved pricing first, fall back to hardcoded discount\n if ri_term:\n ri_map = {\n \"1yr_no_upfront\": (\"1yr\", \"No Upfront\"),\n \"1yr_partial_upfront\": (\"1yr\", \"Partial Upfront\"),\n \"1yr_all_upfront\": (\"1yr\", \"All Upfront\"),\n \"3yr_no_upfront\": (\"3yr\", \"No Upfront\"),\n \"3yr_partial_upfront\": (\"3yr\", \"Partial Upfront\"),\n \"3yr_all_upfront\": (\"3yr\", \"All Upfront\"),\n }\n if ri_term in ri_map:\n lease, purchase = ri_map[ri_term]\n try:\n eff_monthly = loader.get_reserved_effective_monthly(\n region, node_type, engine, lease, purchase\n )\n ri_total = eff_monthly * node_count\n discount_pct = round((1 - ri_total / total) * 100)\n result[\"reserved\"] = {\n \"term\": ri_term,\n \"discount_pct\": discount_pct,\n \"per_node_monthly\": eff_monthly,\n \"total_monthly\": round(ri_total, 2),\n \"annual_savings\": round((total - ri_total) * 12, 2),\n \"source\": \"live\",\n }\n except ValueError:\n pass # no reserved pricing available for this combo\n\n return result\n\n\ndef format_ri_options(loader, engine, region, node_type, node_count):\n \"\"\"Show all reserved pricing options for a specific node configuration.\"\"\"\n options = loader.list_reserved_options(region, node_type, engine)\n hourly = loader.get_node_hourly_rate(region, node_type, engine)\n od_monthly = hourly * HOURS_PER_MONTH * node_count\n\n lines = []\n lines.append(\"=\" * 68)\n lines.append(\"Reserved Pricing Options (live pricing)\")\n lines.append(\"=\" * 68)\n lines.append(\"\")\n lines.append(\" Node type: {} x {} ({} engine)\".format(node_type, node_count, engine))\n lines.append(\" On-demand: ${:,.2f}/month\".format(od_monthly))\n lines.append(\"\")\n\n if not options:\n lines.append(\" No reserved pricing options found for this configuration.\")\n lines.append(\"\")\n lines.append(PRICING_DISCLAIMER)\n lines.append(\"\")\n return \"\\n\".join(lines)\n\n lines.append(\" {:\u003c10} {:\u003c18} {:>12} {:>10} {:>10}\".format(\n \"Term\", \"Purchase Option\", \"Monthly\", \"Upfront\", \"Savings\"))\n lines.append(\" {} {} {} {} {}\".format(\"-\" * 10, \"-\" * 18, \"-\" * 12, \"-\" * 10, \"-\" * 10))\n\n for opt in options:\n eff_total = opt[\"effective_monthly\"] * node_count\n savings_pct = round((1 - eff_total / od_monthly) * 100)\n upfront_total = opt[\"upfront\"] * node_count\n lines.append(\" {:\u003c10} {:\u003c18} ${:>9,.2f} ${:>8,.0f} {:>8}%\".format(\n opt[\"lease_length\"], opt[\"purchase_option\"],\n eff_total, upfront_total, savings_pct))\n\n lines.append(\"\")\n lines.append(PRICING_DISCLAIMER)\n lines.append(\"\")\n return \"\\n\".join(lines)\n\n\ndef estimate_extended_support(loader, engine, region, node_type, node_count):\n \"\"\"Estimate Extended Support surcharge for EOL Redis versions.\"\"\"\n results = {\"engine\": engine, \"region\": region, \"node_type\": node_type, \"node_count\": node_count, \"years\": []}\n\n if not loader.has_extended_support_pricing():\n results[\"note\"] = \"Extended Support pricing not available in bulk pricing API. Check the ElastiCache pricing page.\"\n return results\n\n for year in [\"1\", \"2\", \"3\"]:\n try:\n hourly = loader.get_extended_support_rate(region, node_type, engine, year)\n monthly = hourly * HOURS_PER_MONTH * node_count\n results[\"years\"].append({\n \"year\": int(year),\n \"per_node_hourly\": hourly,\n \"total_monthly\": round(monthly, 2),\n \"total_annual\": round(monthly * 12, 2),\n })\n except ValueError:\n pass\n\n if results[\"years\"]:\n try:\n base_hourly = loader.get_node_hourly_rate(region, node_type, engine)\n base_monthly = base_hourly * HOURS_PER_MONTH * node_count\n results[\"base_monthly\"] = round(base_monthly, 2)\n total_yr1 = base_monthly + results[\"years\"][0][\"total_monthly\"]\n results[\"total_with_surcharge_monthly\"] = round(total_yr1, 2)\n except ValueError:\n pass\n results[\"recommendation\"] = (\n \"Migrating to Valkey eliminates Extended Support charges. \"\n \"Savings vs other engines: 20% lower on node-based, 33% lower on serverless.\"\n )\n\n return results\n\n\ndef format_extended_support_report(results):\n \"\"\"Generate a human-readable Extended Support cost report.\"\"\"\n lines = []\n lines.append(\"=\" * 60)\n lines.append(\"Extended Support Cost Estimate (live pricing)\")\n lines.append(\"=\" * 60)\n lines.append(\"\")\n lines.append(\" Engine: {}\".format(results[\"engine\"]))\n lines.append(\" Region: {}\".format(results[\"region\"]))\n lines.append(\" Node type: {}\".format(results.get(\"node_type\", \"unknown\")))\n lines.append(\" Nodes: {}\".format(results[\"node_count\"]))\n lines.append(\"\")\n\n if not results[\"years\"]:\n lines.append(\" {}\".format(results.get(\"note\", \"No pricing data available.\")))\n lines.append(\"\")\n return \"\\n\".join(lines)\n\n lines.append(\" {:\u003c18} {:>16} {:>16}\".format(\"Year After EOL\", \"Monthly Cost\", \"Annual Cost\"))\n lines.append(\" {} {} {}\".format(\"-\" * 18, \"-\" * 16, \"-\" * 16))\n\n for y in results[\"years\"]:\n lines.append(\" Year {:\u003c13} ${:>13,.2f} ${:>13,.2f}\".format(\n y[\"year\"], y[\"total_monthly\"], y[\"total_annual\"]\n ))\n\n if results.get(\"base_monthly\"):\n lines.append(\"\")\n lines.append(\" Base node cost: ${:>13,.2f}/month\".format(results[\"base_monthly\"]))\n lines.append(\" Total (base + Yr1): ${:>13,.2f}/month\".format(results[\"total_with_surcharge_monthly\"]))\n\n lines.append(\"\")\n if \"recommendation\" in results:\n lines.append(\" {}\".format(results[\"recommendation\"]))\n lines.append(\"\")\n lines.append(PRICING_DISCLAIMER)\n lines.append(\"\")\n return \"\\n\".join(lines)\n\n\ndef estimate_ecpu_from_ops(ops_per_sec, avg_ecpu_per_op=1.0):\n \"\"\"Convert operations per second to monthly ECPUs (millions).\n\n Default: 1 ECPU per operation (assumes simple GET/SET with payload under 1 KB).\n For larger payloads, use --avg-payload-kb which scales ECPUs linearly.\n For complex commands (SORT, ZADD, etc.), pass a higher avg_ecpu_per_op.\n \"\"\"\n ecpu_per_month = ops_per_sec * avg_ecpu_per_op * 3600 * HOURS_PER_MONTH\n return round(ecpu_per_month / 1_000_000, 1)\n\n\ndef interactive(loader, region):\n \"\"\"Interactive interview-style cost estimation.\"\"\"\n print(\"=== ElastiCache Price Calculator (live pricing) ===\\n\")\n\n engine = input(\"Engine [valkey/redis/memcached] (default: valkey): \").strip().lower() or \"valkey\"\n data_gb = float(input(\"Estimated data size in GB (default: 1): \").strip() or \"1\")\n mode = input(\"Deployment [serverless/node] (default: serverless): \").strip().lower() or \"serverless\"\n\n if mode == \"serverless\":\n ops = input(\"Estimated operations per second (default: 100): \").strip() or \"100\"\n ops_per_sec = float(ops)\n ecpu_millions = estimate_ecpu_from_ops(ops_per_sec)\n print(\" -> Estimated {}M ECPUs/month\".format(ecpu_millions))\n result = estimate_serverless(loader, engine, region, data_gb, ecpu_millions)\n print(\"\\n{}\".format(json.dumps(result, indent=2)))\n\n elif mode == \"node\":\n node_type = input(\"Node type (default: cache.r7g.large): \").strip() or \"cache.r7g.large\"\n nodes = int(input(\"Number of nodes (default: 2): \").strip() or \"2\")\n result = estimate_node_based(loader, engine, region, node_type, nodes, ri_term=\"1yr_no_upfront\")\n print(\"\\n{}\".format(json.dumps(result, indent=2)))\n show_ri = input(\"\\nShow all reserved options? [y/n] (default: n): \").strip().lower()\n if show_ri == \"y\":\n print(\"\\n{}\".format(format_ri_options(loader, engine, region, node_type, nodes)))\n\n\nif __name__ == \"__main__\":\n parser = argparse.ArgumentParser(description=\"ElastiCache Price Calculator (live pricing)\")\n parser.add_argument(\"--interactive\", action=\"store_true\", help=\"Interactive mode\")\n parser.add_argument(\"--engine\", choices=[\"valkey\", \"redis\", \"memcached\"], default=\"valkey\")\n parser.add_argument(\"--mode\", choices=[\"serverless\", \"node\"], default=\"serverless\")\n parser.add_argument(\"--region\", default=REGION, help=\"AWS region (default: from env or us-east-1)\")\n parser.add_argument(\"--data-gb\", type=float, default=1.0, help=\"Data stored in GB\")\n parser.add_argument(\"--ecpu-millions\", type=float, default=None, help=\"ECPUs per month (millions)\")\n parser.add_argument(\"--ops-per-sec\", type=float, default=None, help=\"Operations per second (auto-converts to ECPUs)\")\n parser.add_argument(\"--node-type\", default=\"cache.r7g.large\", help=\"Node type for node-based\")\n parser.add_argument(\"--nodes\", type=int, default=2, help=\"Node count for node-based\")\n parser.add_argument(\"--ri-term\", default=None,\n choices=[\"1yr_no_upfront\", \"1yr_partial_upfront\", \"1yr_all_upfront\",\n \"3yr_no_upfront\", \"3yr_partial_upfront\", \"3yr_all_upfront\"],\n help=\"Reserved instance term\")\n parser.add_argument(\"--show-ri-options\", action=\"store_true\",\n help=\"Show all reserved pricing options for the node type\")\n parser.add_argument(\"--extended-support\", action=\"store_true\",\n help=\"Show Extended Support surcharge for EOL Redis versions\")\n parser.add_argument(\"--avg-payload-kb\", type=float, default=1.0,\n help=\"Average payload size in KB (default: 1.0). ECPUs scale linearly with payload.\")\n parser.add_argument(\"--pricing-csv\", default=None, help=\"Optional local pricing CSV (overrides live fetch)\")\n args = parser.parse_args()\n\n loader = PricingLoader(args.pricing_csv)\n\n if args.interactive:\n interactive(loader, args.region)\n sys.exit(0)\n\n # Auto-convert ops/sec to ECPUs if provided\n ecpu_millions = args.ecpu_millions\n if ecpu_millions is None and args.ops_per_sec:\n ecpu_millions = estimate_ecpu_from_ops(args.ops_per_sec)\n elif ecpu_millions is None:\n ecpu_millions = 100.0\n\n if args.extended_support:\n result = estimate_extended_support(loader, args.engine, args.region, args.node_type, args.nodes)\n print(format_extended_support_report(result))\n elif args.show_ri_options:\n print(format_ri_options(loader, args.engine, args.region, args.node_type, args.nodes))\n elif args.mode == \"serverless\":\n result = estimate_serverless(loader, args.engine, args.region, args.data_gb, ecpu_millions, args.avg_payload_kb)\n result[\"disclaimer\"] = \"Estimates only. Verify at https://aws.amazon.com/elasticache/pricing/\"\n print(json.dumps(result, indent=2))\n elif args.mode == \"node\":\n result = estimate_node_based(loader, args.engine, args.region, args.node_type, args.nodes, ri_term=args.ri_term)\n result[\"disclaimer\"] = \"Estimates only. Verify at https://aws.amazon.com/elasticache/pricing/\"\n print(json.dumps(result, indent=2))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":14744,"content_sha256":"5f0a76080489fde596080bd5e210a440d2362c41bb8060d9f9a0735ca6c2e99e"},{"filename":"scripts/pricing.py","content":"\"\"\"\nElastiCache pricing loader.\n\nFetches live pricing from the AWS Bulk Pricing API (public, no auth required).\nCaches locally to avoid repeated downloads. Refreshes if cache is older than 7 days.\n\nSource: https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonElastiCache/current/index.json\n\nAlso accepts an optional local CSV for offline or override use.\n\nSupports On-Demand and Reserved Node pricing (1-year and 3-year terms with\nNo Upfront, Partial Upfront, and All Upfront purchase options). Reserved pricing\nis extracted from the bulk pricing JSON terms.Reserved section.\n\"\"\"\nimport csv\nimport json\nimport os\nimport sys\nimport time\nimport urllib.request\nfrom typing import Dict, Optional, Tuple\n\nBULK_PRICING_URL = \"https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonElastiCache/current/index.json\"\nCACHE_MAX_AGE_SECONDS = 7 * 24 * 3600 # 7 days\n_CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), \".pricing_cache\")\n_CACHE_FILE = os.path.join(_CACHE_DIR, \"elasticache_pricing.json\")\n\n\ndef _cache_is_fresh() -> bool:\n \"\"\"Check if local cache exists and is younger than CACHE_MAX_AGE_SECONDS.\"\"\"\n if not os.path.exists(_CACHE_FILE):\n return False\n age = time.time() - os.path.getmtime(_CACHE_FILE)\n return age \u003c CACHE_MAX_AGE_SECONDS\n\n\ndef _fetch_and_cache() -> dict:\n \"\"\"Fetch bulk pricing JSON from AWS and cache locally.\"\"\"\n os.makedirs(_CACHE_DIR, exist_ok=True)\n print(\"Fetching live ElastiCache pricing from AWS...\", file=sys.stderr)\n try:\n req = urllib.request.Request(BULK_PRICING_URL)\n with urllib.request.urlopen(req, timeout=60) as resp:\n raw = resp.read()\n except Exception as e:\n raise RuntimeError(\n \"Failed to fetch pricing from {}. Error: {}. \"\n \"Check network connectivity or provide a local pricing CSV.\".format(BULK_PRICING_URL, e)\n )\n\n data = json.loads(raw)\n parsed = _parse_bulk_pricing(data)\n parsed[\"fetched_at\"] = time.time()\n\n with open(_CACHE_FILE, \"w\") as f:\n json.dump(parsed, f)\n\n print(\" Pricing cached at: {}\".format(_CACHE_FILE), file=sys.stderr)\n return parsed\n\n\ndef _load_cache() -> dict:\n \"\"\"Load parsed pricing from local cache.\"\"\"\n with open(_CACHE_FILE, \"r\") as f:\n return json.load(f)\n\n\ndef _parse_bulk_pricing(data: dict) -> dict:\n \"\"\"Parse the AWS bulk pricing JSON into our compact format.\n\n Extracts:\n - On-demand node-based hourly rates by (region, instance_type, engine)\n - Reserved node-based rates by (region, instance_type, engine, lease, purchase_option)\n - Serverless storage rates by (region, engine)\n - Serverless ECPU rates by (region, engine)\n - Extended Support surcharges by (region, engine, year)\n \"\"\"\n products = data.get(\"products\", {})\n terms = data.get(\"terms\", {}).get(\"OnDemand\", {})\n reserved_terms = data.get(\"terms\", {}).get(\"Reserved\", {})\n\n node_pricing = {} # \"region|instance_type|engine\" -> $/hr\n serverless_storage = {} # \"region|engine\" -> $/GB-hr\n serverless_ecpu = {} # \"region|engine\" -> $/ECPU\n extended_support = {} # \"region|instance_type|engine|year\" -> $/node-hr\n reserved_pricing = {} # \"region|instance_type|engine|lease|purchase\" -> {\"hourly\": $, \"upfront\": $}\n node_memory = {} # \"instance_type\" -> GB (float)\n\n for sku, product in products.items():\n attrs = product.get(\"attributes\", {})\n family = product.get(\"productFamily\", \"\")\n region = attrs.get(\"regionCode\", \"\")\n if not region:\n continue\n # Skip Outpost and non-standard location types\n if attrs.get(\"locationType\", \"\") != \"AWS Region\":\n continue\n\n # Get the on-demand price and unit for this SKU\n price_info = _get_od_price_with_unit(sku, terms)\n if price_info is None:\n continue\n price_per_unit, unit = price_info\n if price_per_unit \u003c= 0:\n continue\n\n if family == \"Cache Instance\":\n instance_type = attrs.get(\"instanceType\", \"\")\n engine = _normalize_engine(attrs.get(\"cacheEngine\", \"\"))\n usagetype = attrs.get(\"usagetype\", \"\")\n\n # Extended Support entries live under \"Cache Instance\" family\n # but have \"ExtendedSupport\" in the usagetype\n if \"ExtendedSupport\" in usagetype:\n if not engine or not instance_type or unit != \"Hrs\":\n continue\n years = _parse_extended_support_years(usagetype)\n for year in years:\n key = \"{}|{}|{}|{}\".format(region, instance_type, engine, year)\n extended_support[key] = price_per_unit\n elif instance_type and engine and unit == \"Hrs\":\n key = \"{}|{}|{}\".format(region, instance_type, engine)\n node_pricing[key] = price_per_unit\n # Extract memory from product attributes (e.g., \"103.68 GiB\")\n if instance_type not in node_memory:\n mem_str = attrs.get(\"memory\", \"\")\n mem_gb = _parse_memory_gib(mem_str)\n if mem_gb > 0:\n node_memory[instance_type] = mem_gb\n\n elif family == \"ElastiCache Serverless\":\n engine = _normalize_engine(attrs.get(\"cacheEngine\", \"\"))\n operation = attrs.get(\"operation\", \"\")\n usagetype = attrs.get(\"usagetype\", \"\")\n if \"Snapshot\" in operation or \"Backup\" in usagetype:\n continue\n if not engine:\n continue\n key = \"{}|{}\".format(region, engine)\n unit_lower = unit.lower()\n if \"gb-hour\" in unit_lower or \"CachedData\" in usagetype:\n serverless_storage[key] = price_per_unit\n elif \"processingunit\" in unit_lower or \"ProcessingUnits\" in usagetype:\n serverless_ecpu[key] = price_per_unit\n\n # Parse Reserved terms for node-based instances\n _valid_purchase_options = {\"No Upfront\", \"Partial Upfront\", \"All Upfront\"}\n for sku, offers in reserved_terms.items():\n product = products.get(sku)\n if not product or product.get(\"productFamily\") != \"Cache Instance\":\n continue\n attrs = product.get(\"attributes\", {})\n if attrs.get(\"locationType\", \"\") != \"AWS Region\":\n continue\n if \"ExtendedSupport\" in attrs.get(\"usagetype\", \"\"):\n continue\n region = attrs.get(\"regionCode\", \"\")\n instance_type = attrs.get(\"instanceType\", \"\")\n engine = _normalize_engine(attrs.get(\"cacheEngine\", \"\"))\n if not (region and instance_type and engine):\n continue\n\n for offer_id, offer in offers.items():\n ta = offer.get(\"termAttributes\", {})\n purchase_option = ta.get(\"PurchaseOption\", \"\")\n if purchase_option not in _valid_purchase_options:\n continue # skip legacy \"Heavy Utilization\"\n lease = ta.get(\"LeaseContractLength\", \"\") # \"1yr\" or \"3yr\"\n\n hourly = 0.0\n upfront = 0.0\n for dim in offer.get(\"priceDimensions\", {}).values():\n try:\n price = float(dim.get(\"pricePerUnit\", {}).get(\"USD\", \"0\"))\n except (ValueError, TypeError):\n continue\n unit = dim.get(\"unit\", \"\")\n if unit == \"Hrs\":\n hourly = price\n elif unit == \"Quantity\":\n upfront = price\n\n key = \"{}|{}|{}|{}|{}\".format(\n region, instance_type, engine, lease, purchase_option\n )\n reserved_pricing[key] = {\"hourly\": hourly, \"upfront\": upfront}\n\n return {\n \"node_pricing\": node_pricing,\n \"serverless_storage\": serverless_storage,\n \"serverless_ecpu\": serverless_ecpu,\n \"extended_support\": extended_support,\n \"reserved_pricing\": reserved_pricing,\n \"node_memory\": node_memory,\n }\n\n\ndef _get_od_price_with_unit(sku: str, terms: dict) -> Optional[Tuple[float, str]]:\n \"\"\"Extract on-demand price and unit for a SKU from the terms section.\"\"\"\n sku_terms = terms.get(sku, {})\n for offer in sku_terms.values():\n dimensions = offer.get(\"priceDimensions\", {})\n for dim in dimensions.values():\n try:\n price = float(dim.get(\"pricePerUnit\", {}).get(\"USD\", \"0\"))\n unit = dim.get(\"unit\", \"\")\n return (price, unit)\n except (ValueError, TypeError):\n continue\n return None\n\n\ndef _parse_extended_support_years(usagetype: str) -> list:\n \"\"\"Extract year tiers from an Extended Support usagetype string.\n\n AWS usagetype format:\n - \"USE1-ExtendedSupportYr1_Yr2-NodeUsage:cache.r7g.large\" -> years 1 and 2 (same rate)\n - \"USE1-ExtendedSupportYr3-NodeUsage:cache.r7g.large\" -> year 3+\n\n Returns a list of year strings, e.g. [\"1\", \"2\"] or [\"3\"].\n \"\"\"\n ut = usagetype.lower()\n if \"yr1_yr2\" in ut:\n return [\"1\", \"2\"]\n if \"yr3\" in ut:\n return [\"3\"]\n if \"yr1\" in ut and \"yr2\" not in ut:\n return [\"1\"]\n if \"yr2\" in ut and \"yr1\" not in ut:\n return [\"2\"]\n # Fallback: generic extended support mention without year\n if \"extendedsupport\" in ut.replace(\"-\", \"\").replace(\"_\", \"\"):\n return [\"1\"]\n return []\n\n\ndef _normalize_engine(engine_str: str) -> str:\n \"\"\"Normalize engine name from bulk pricing to our standard names.\"\"\"\n eng = engine_str.lower().strip()\n if \"valkey\" in eng:\n return \"valkey\"\n if \"redis\" in eng:\n return \"redis\"\n if \"memcached\" in eng:\n return \"memcached\"\n return eng\n\n\ndef _parse_memory_gib(mem_str: str) -> float:\n \"\"\"Parse memory string like '103.68 GiB' into a float GB value.\"\"\"\n if not mem_str:\n return 0.0\n try:\n parts = mem_str.strip().split()\n return float(parts[0])\n except (ValueError, IndexError):\n return 0.0\n\n\nclass PricingLoader:\n \"\"\"Load and look up ElastiCache pricing.\n\n Priority:\n 1. Local CSV (if provided) - for offline or override use\n 2. Live AWS bulk pricing (fetched and cached locally for 7 days)\n \"\"\"\n\n def __init__(self, pricing_csv: Optional[str] = None):\n self._serverless_storage: Dict[Tuple[str, str], float] = {}\n self._serverless_ecpu: Dict[Tuple[str, str], float] = {}\n self._node_od: Dict[Tuple[str, str, str], float] = {}\n self._extended_support: Dict[Tuple[str, str, str, str], float] = {} # (region, itype, engine, year)\n self._reserved: Dict[Tuple[str, str, str, str, str], dict] = {} # (region, itype, engine, lease, purchase) -> {\"hourly\", \"upfront\"}\n self._node_memory: Dict[str, float] = {} # instance_type -> GB\n\n if pricing_csv and os.path.exists(pricing_csv):\n self._load_csv(pricing_csv)\n elif pricing_csv:\n raise FileNotFoundError(\n \"Pricing CSV not found: {}\".format(pricing_csv)\n )\n else:\n self._load_live()\n\n def _load_live(self):\n \"\"\"Load pricing from AWS bulk pricing (cached locally).\"\"\"\n if _cache_is_fresh():\n data = _load_cache()\n else:\n data = _fetch_and_cache()\n\n for key, price in data.get(\"node_pricing\", {}).items():\n parts = key.split(\"|\")\n if len(parts) == 3:\n region, itype, engine = parts\n self._node_od[(region, itype, engine)] = price\n\n for key, price in data.get(\"serverless_storage\", {}).items():\n parts = key.split(\"|\")\n if len(parts) == 2:\n region, engine = parts\n self._serverless_storage[(region, engine)] = price\n\n for key, price in data.get(\"serverless_ecpu\", {}).items():\n parts = key.split(\"|\")\n if len(parts) == 2:\n region, engine = parts\n self._serverless_ecpu[(region, engine)] = price\n\n for key, price in data.get(\"extended_support\", {}).items():\n parts = key.split(\"|\")\n if len(parts) == 4:\n region, itype, engine, year = parts\n self._extended_support[(region, itype, engine, year)] = price\n\n for key, price_info in data.get(\"reserved_pricing\", {}).items():\n parts = key.split(\"|\")\n if len(parts) == 5:\n region, itype, engine, lease, purchase = parts\n self._reserved[(region, itype, engine, lease, purchase)] = price_info\n\n self._node_memory = data.get(\"node_memory\", {})\n\n def _load_csv(self, path: str):\n \"\"\"Load pricing from a local CSV.\"\"\"\n with open(path, newline=\"\") as f:\n for row in csv.DictReader(f):\n family = row.get(\"Product Family\", \"\")\n region = row.get(\"Region Code\", \"\")\n engine = row.get(\"Cache Engine\", \"\").lower()\n term = row.get(\"TermType\", \"\")\n try:\n price = float(row.get(\"PricePerUnit\", \"0\"))\n except (ValueError, TypeError):\n continue\n if price \u003c= 0 or not region:\n continue\n\n if family == \"Cache Instance\" and term == \"OnDemand\":\n itype = row.get(\"Instance Type\", \"\")\n if itype and row.get(\"Unit\") == \"Hrs\":\n self._node_od[(region, itype, engine)] = price\n\n elif family == \"ElastiCache Serverless\":\n unit = row.get(\"Unit\", \"\")\n operation = row.get(\"Operation\", row.get(\"operation\", \"\"))\n if \"Snapshot\" in operation or unit == \"GB-months\":\n continue\n unit_lower = unit.lower()\n if \"gb-hour\" in unit_lower:\n self._serverless_storage[(region, engine)] = price\n elif \"processingunit\" in unit_lower:\n self._serverless_ecpu[(region, engine)] = price\n\n # CSV loader only supports on-demand node and serverless pricing.\n # Reserved pricing, Extended Support surcharges, and node memory data\n # are only available from the live bulk pricing API.\n if not self._reserved:\n print(\n \"Note: CSV pricing does not include Reserved Instance or Extended Support data. \"\n \"Use live pricing (omit --pricing-csv) for full coverage.\",\n file=sys.stderr,\n )\n\n def get_serverless_storage_rate(self, region: str, engine: str) -> float:\n \"\"\"Get serverless storage rate in $/GB-hour.\"\"\"\n eng = engine.lower()\n key = (region, eng)\n if key in self._serverless_storage:\n return self._serverless_storage[key]\n raise ValueError(\n \"No serverless storage pricing found for region={}, engine={}. \"\n \"Available: {}\".format(region, eng, self._list_serverless_regions(eng))\n )\n\n def get_serverless_ecpu_rate(self, region: str, engine: str) -> float:\n \"\"\"Get serverless ECPU rate in $ per single ECPU.\"\"\"\n eng = engine.lower()\n key = (region, eng)\n if key in self._serverless_ecpu:\n return self._serverless_ecpu[key]\n raise ValueError(\n \"No serverless ECPU pricing found for region={}, engine={}. \"\n \"Available: {}\".format(region, eng, self._list_serverless_regions(eng))\n )\n\n def get_node_hourly_rate(self, region: str, instance_type: str, engine: str = \"valkey\") -> float:\n \"\"\"Get node on-demand hourly rate.\"\"\"\n eng = engine.lower()\n key = (region, instance_type, eng)\n if key in self._node_od:\n return self._node_od[key]\n raise ValueError(\n \"No node pricing found for region={}, instance_type={}, engine={}\".format(\n region, instance_type, eng\n )\n )\n\n def get_node_monthly_rate(self, region: str, instance_type: str, engine: str = \"valkey\") -> float:\n \"\"\"Get node on-demand monthly rate (hourly x 730 hours).\"\"\"\n return self.get_node_hourly_rate(region, instance_type, engine) * 730\n\n def get_extended_support_rate(self, region: str, instance_type: str, engine: str, year: str = \"1\") -> float:\n \"\"\"Get Extended Support surcharge in $/node-hour for the given year tier.\n\n Year tiers: '1' and '2' (same rate, from Yr1_Yr2), '3' (higher rate, from Yr3).\n \"\"\"\n eng = engine.lower()\n key = (region, instance_type, eng, str(year))\n if key in self._extended_support:\n return self._extended_support[key]\n raise ValueError(\n \"No Extended Support pricing found for region={}, instance_type={}, engine={}, year={}. \"\n \"This may mean Extended Support pricing is not published for this combination.\".format(\n region, instance_type, eng, year\n )\n )\n\n def has_extended_support_pricing(self) -> bool:\n \"\"\"Check if any Extended Support pricing was loaded.\"\"\"\n return len(self._extended_support) > 0\n\n def get_node_memory_gb(self, instance_type: str) -> float:\n \"\"\"Get memory capacity in GB for a node type. Returns 0 if unknown.\"\"\"\n return self._node_memory.get(instance_type, 0.0)\n\n def get_all_node_memory(self) -> dict:\n \"\"\"Get the full instance_type -> memory_gb mapping.\"\"\"\n return dict(self._node_memory)\n\n def _list_serverless_regions(self, engine: str) -> str:\n \"\"\"List regions that have serverless pricing loaded for an engine.\"\"\"\n regions = sorted(set(\n r for (r, e) in self._serverless_storage if e == engine\n ))\n if len(regions) > 5:\n return \", \".join(regions[:5]) + \" and {} more\".format(len(regions) - 5)\n return \", \".join(regions) if regions else \"none loaded\"\n\n # --- Reserved pricing methods ---\n\n def get_reserved_rate(\n self, region: str, instance_type: str, engine: str = \"valkey\",\n lease_length: str = \"1yr\", purchase_option: str = \"No Upfront\",\n ) -> dict:\n \"\"\"Get reserved pricing for a node.\n\n Returns {\"hourly\": float, \"upfront\": float}.\n - \"No Upfront\": upfront=0, hourly=full reserved rate\n - \"All Upfront\": upfront=lump sum, hourly=0\n - \"Partial Upfront\": upfront=partial lump sum, hourly=reduced rate\n \"\"\"\n eng = engine.lower()\n key = (region, instance_type, eng, lease_length, purchase_option)\n if key in self._reserved:\n return self._reserved[key]\n raise ValueError(\n \"No reserved pricing found for region={}, instance_type={}, engine={}, \"\n \"lease={}, purchase={}\".format(region, instance_type, eng, lease_length, purchase_option)\n )\n\n def get_reserved_effective_monthly(\n self, region: str, instance_type: str, engine: str = \"valkey\",\n lease_length: str = \"1yr\", purchase_option: str = \"No Upfront\",\n ) -> float:\n \"\"\"Get effective monthly cost for a reserved node.\n\n Amortizes upfront payment over the term and adds the recurring hourly cost.\n Formula: (upfront / term_months) + (hourly * 730)\n \"\"\"\n rate = self.get_reserved_rate(region, instance_type, engine, lease_length, purchase_option)\n term_months = 12 if lease_length == \"1yr\" else 36\n monthly = (rate[\"upfront\"] / term_months) + (rate[\"hourly\"] * 730)\n return round(monthly, 2)\n\n def list_reserved_options(\n self, region: str, instance_type: str, engine: str = \"valkey\",\n ) -> list:\n \"\"\"List all available reserved pricing options for a node.\n\n Returns a list of dicts with lease_length, purchase_option,\n hourly, upfront, and effective_monthly.\n \"\"\"\n eng = engine.lower()\n results = []\n for (r, it, e, lease, purchase), rate in self._reserved.items():\n if r == region and it == instance_type and e == eng:\n term_months = 12 if lease == \"1yr\" else 36\n effective = round(\n (rate[\"upfront\"] / term_months) + (rate[\"hourly\"] * 730), 2\n )\n results.append({\n \"lease_length\": lease,\n \"purchase_option\": purchase,\n \"hourly\": rate[\"hourly\"],\n \"upfront\": rate[\"upfront\"],\n \"effective_monthly\": effective,\n })\n results.sort(key=lambda x: x[\"effective_monthly\"])\n return results\n","content_type":"text/x-python; charset=utf-8","language":"python","size":20645,"content_sha256":"d7dbf9ef87af02149b43ca1153b85b870bee6dcfca0127076708a0d0feb9a149"},{"filename":"scripts/security_audit.py","content":"#!/usr/bin/env python3\n\"\"\"\nElastiCache Security Audit\n\nInspects an existing cache and reports its security and operational posture.\nChecks: TLS, auth mode, encryption at rest, Multi-AZ, failover, backups,\nsecurity group openness, and engine currency.\n\nUsage:\n # Audit a serverless cache\n python security_audit.py --serverless my-cache --region us-east-1\n\n # Audit a node-based replication group\n python security_audit.py --replication-group my-cluster --region us-east-1\n\n # JSON output for automation\n python security_audit.py --serverless my-cache --region us-east-1 --json\n\nRequires: boto3, AWS credentials with elasticache:Describe*,\n elasticache:ListTagsForResource, and ec2:DescribeSecurityGroups\n\"\"\"\n\nimport argparse\nimport json\nimport sys\n\n# ---------------------------------------------------------------------------\n# Dependency check -- fail early with an actionable message instead of a\n# raw ImportError traceback.\n# ---------------------------------------------------------------------------\ntry:\n import boto3\n from botocore.exceptions import ClientError, BotoCoreError\nexcept ImportError:\n print(\n \"Error: the 'boto3' package is required for this script.\\n\"\n \"Install it with:\\n\"\n \" pip install boto3\\n\"\n \"\\n\"\n \"You also need AWS credentials with elasticache:Describe*,\\n\"\n \"elasticache:ListTagsForResource, and ec2:DescribeSecurityGroups\\n\"\n \"permissions configured (via environment\\n\"\n \"variables, ~/.aws/credentials, or an IAM role).\"\n )\n sys.exit(1)\n\n\ndef audit_serverless(client, ec2_client, cache_name):\n \"\"\"Audit a serverless cache.\"\"\"\n try:\n resp = client.describe_serverless_caches(ServerlessCacheName=cache_name)\n except client.exceptions.ServerlessCacheNotFoundFault:\n return {\"error\": f\"Serverless cache '{cache_name}' not found\"}\n except Exception as e:\n if hasattr(e, 'response') and e.response.get('Error', {}).get('Code') == 'ServerlessCacheNotFoundFault':\n return {\"error\": f\"Serverless cache '{cache_name}' not found\"}\n raise\n caches = resp.get(\"ServerlessCaches\", [])\n if not caches:\n return {\"error\": f\"Serverless cache '{cache_name}' not found\"}\n\n cache = caches[0]\n findings = []\n info = {\n \"cache_type\": \"serverless\",\n \"cache_name\": cache_name,\n \"engine\": cache.get(\"Engine\", \"unknown\"),\n \"engine_version\": cache.get(\"MajorEngineVersion\", \"unknown\"),\n \"status\": cache.get(\"Status\", \"unknown\"),\n \"arn\": cache.get(\"ARN\", \"unknown\"),\n }\n\n # TLS: always on for serverless\n info[\"tls_enabled\"] = True\n findings.append({\"check\": \"TLS in-transit encryption\", \"status\": \"PASS\", \"detail\": \"Always enabled on serverless\"})\n\n # Auth: serverless requires RBAC\n user_group = cache.get(\"UserGroupId\")\n if user_group:\n findings.append({\"check\": \"Authentication\", \"status\": \"PASS\", \"detail\": f\"RBAC user group: {user_group}\"})\n info[\"auth_mode\"] = \"RBAC\"\n info[\"user_group\"] = user_group\n else:\n findings.append({\"check\": \"Authentication\", \"status\": \"WARN\", \"detail\": \"No user group attached — using default user only\"})\n info[\"auth_mode\"] = \"default-user-only\"\n\n # Encryption at rest: check KMS\n kms_key = cache.get(\"KmsKeyId\")\n if kms_key:\n findings.append({\"check\": \"Encryption at rest\", \"status\": \"PASS\", \"detail\": f\"KMS key: {kms_key}\"})\n info[\"encryption_at_rest\"] = True\n info[\"kms_key\"] = kms_key\n else:\n findings.append({\"check\": \"Encryption at rest\", \"status\": \"PASS\", \"detail\": \"Always enabled on serverless (using AWS-managed key)\"})\n info[\"encryption_at_rest\"] = True\n\n # Multi-AZ: always on for serverless\n info[\"multi_az\"] = True\n findings.append({\"check\": \"Multi-AZ\", \"status\": \"PASS\", \"detail\": \"Always enabled on serverless\"})\n\n # Backups: check SnapshotRetentionLimit\n retention = cache.get(\"SnapshotRetentionLimit\", 0)\n info[\"snapshot_retention_days\"] = retention\n if retention and retention > 0:\n snapshot_time = cache.get(\"DailySnapshotTime\", \"not set\")\n info[\"daily_backups\"] = True\n findings.append({\"check\": \"Automatic backups\", \"status\": \"PASS\", \"detail\": f\"Retention: {retention} days, daily snapshot time: {snapshot_time}\"})\n else:\n info[\"daily_backups\"] = False\n findings.append({\"check\": \"Automatic backups\", \"status\": \"FAIL\", \"detail\": \"No automatic backups configured, SnapshotRetentionLimit is 0 or not set\"})\n\n # Usage limits (cost control)\n limits = cache.get(\"CacheUsageLimits\", {})\n data_limit = limits.get(\"DataStorage\", {})\n ecpu_limit = limits.get(\"ECPUPerSecond\", {})\n if data_limit.get(\"Maximum\"):\n info[\"max_data_gb\"] = data_limit[\"Maximum\"]\n findings.append({\"check\": \"Data storage limit\", \"status\": \"PASS\", \"detail\": f\"{data_limit['Maximum']} {data_limit.get('Unit', 'GB')} max\"})\n else:\n findings.append({\"check\": \"Data storage limit\", \"status\": \"WARN\", \"detail\": \"No data storage limit set — costs could grow unbounded\"})\n if ecpu_limit.get(\"Maximum\"):\n info[\"max_ecpu_per_sec\"] = ecpu_limit[\"Maximum\"]\n findings.append({\"check\": \"ECPU limit\", \"status\": \"PASS\", \"detail\": f\"{ecpu_limit['Maximum']} ECPUs/sec max\"})\n else:\n findings.append({\"check\": \"ECPU limit\", \"status\": \"WARN\", \"detail\": \"No ECPU limit set — costs could grow unbounded\"})\n\n # Security groups\n sg_ids = cache.get(\"SecurityGroupIds\", [])\n sg_findings = _audit_security_groups(ec2_client, sg_ids)\n findings.extend(sg_findings)\n info[\"security_groups\"] = sg_ids\n\n # Tags\n try:\n tag_resp = client.list_tags_for_resource(ResourceName=cache[\"ARN\"])\n tags = {t[\"Key\"]: t[\"Value\"] for t in tag_resp.get(\"TagList\", [])}\n except (ClientError, BotoCoreError):\n tags = {}\n _audit_tags(findings, tags)\n info[\"tags\"] = tags\n\n return {\"info\": info, \"findings\": findings}\n\n\ndef audit_replication_group(client, ec2_client, rg_id):\n \"\"\"Audit a node-based replication group.\"\"\"\n try:\n resp = client.describe_replication_groups(ReplicationGroupId=rg_id)\n except client.exceptions.ReplicationGroupNotFoundFault:\n return {\"error\": f\"Replication group '{rg_id}' not found\"}\n except Exception as e:\n if hasattr(e, 'response') and e.response.get('Error', {}).get('Code') == 'ReplicationGroupNotFoundFault':\n return {\"error\": f\"Replication group '{rg_id}' not found\"}\n raise\n groups = resp.get(\"ReplicationGroups\", [])\n if not groups:\n return {\"error\": f\"Replication group '{rg_id}' not found\"}\n\n rg = groups[0]\n findings = []\n info = {\n \"cache_type\": \"node-based\",\n \"replication_group_id\": rg_id,\n \"engine\": \"unknown\",\n \"status\": rg.get(\"Status\", \"unknown\"),\n \"arn\": rg.get(\"ARN\", \"unknown\"),\n \"cluster_enabled\": rg.get(\"ClusterEnabled\", False),\n \"num_node_groups\": len(rg.get(\"NodeGroups\", [])),\n }\n\n # Engine version from member clusters\n member_clusters = rg.get(\"MemberClusters\", [])\n if member_clusters:\n try:\n cluster_resp = client.describe_cache_clusters(CacheClusterId=member_clusters[0])\n cluster = cluster_resp[\"CacheClusters\"][0]\n info[\"engine\"] = cluster.get(\"Engine\", \"unknown\")\n info[\"engine_version\"] = cluster.get(\"EngineVersion\", \"unknown\")\n info[\"node_type\"] = cluster.get(\"CacheNodeType\", \"unknown\")\n except (ClientError, BotoCoreError):\n info[\"engine_version\"] = \"unknown\"\n info[\"node_type\"] = \"unknown\"\n\n # TLS\n tls = rg.get(\"TransitEncryptionEnabled\", False)\n info[\"tls_enabled\"] = tls\n if tls:\n findings.append({\"check\": \"TLS in-transit encryption\", \"status\": \"PASS\", \"detail\": \"Enabled\"})\n else:\n findings.append({\"check\": \"TLS in-transit encryption\", \"status\": \"FAIL\", \"detail\": \"Not enabled — data in transit is unencrypted\"})\n\n # Encryption at rest\n at_rest = rg.get(\"AtRestEncryptionEnabled\", False)\n info[\"encryption_at_rest\"] = at_rest\n if at_rest:\n kms = rg.get(\"KmsKeyId\", \"AWS-managed\")\n findings.append({\"check\": \"Encryption at rest\", \"status\": \"PASS\", \"detail\": f\"Enabled (key: {kms})\"})\n info[\"kms_key\"] = kms\n else:\n findings.append({\"check\": \"Encryption at rest\", \"status\": \"FAIL\", \"detail\": \"Not enabled — data at rest is unencrypted\"})\n\n # Auth\n auth_token = rg.get(\"AuthTokenEnabled\", False)\n user_group_ids = rg.get(\"UserGroupIds\", [])\n if user_group_ids:\n info[\"auth_mode\"] = \"RBAC\"\n info[\"user_groups\"] = user_group_ids\n findings.append({\"check\": \"Authentication\", \"status\": \"PASS\", \"detail\": f\"RBAC user groups: {', '.join(user_group_ids)}\"})\n elif auth_token:\n info[\"auth_mode\"] = \"AUTH-token\"\n findings.append({\"check\": \"Authentication\", \"status\": \"WARN\", \"detail\": \"Using legacy AUTH token — consider migrating to RBAC\"})\n else:\n info[\"auth_mode\"] = \"none\"\n findings.append({\"check\": \"Authentication\", \"status\": \"FAIL\", \"detail\": \"No authentication configured — any VPC client can access data\"})\n\n # Multi-AZ\n multi_az = rg.get(\"MultiAZ\", \"disabled\")\n info[\"multi_az\"] = multi_az == \"enabled\"\n if multi_az == \"enabled\":\n findings.append({\"check\": \"Multi-AZ\", \"status\": \"PASS\", \"detail\": \"Enabled\"})\n else:\n findings.append({\"check\": \"Multi-AZ\", \"status\": \"WARN\", \"detail\": \"Not enabled — single-AZ failure risk\"})\n\n # Automatic failover\n failover = rg.get(\"AutomaticFailover\", \"disabled\")\n info[\"automatic_failover\"] = failover == \"enabled\"\n if failover == \"enabled\":\n findings.append({\"check\": \"Automatic failover\", \"status\": \"PASS\", \"detail\": \"Enabled\"})\n else:\n findings.append({\"check\": \"Automatic failover\", \"status\": \"WARN\", \"detail\": \"Not enabled — manual intervention needed on primary failure\"})\n\n # Backups\n retention = rg.get(\"SnapshotRetentionLimit\", 0)\n info[\"snapshot_retention_days\"] = retention\n if retention > 0:\n findings.append({\"check\": \"Automatic backups\", \"status\": \"PASS\", \"detail\": f\"Retention: {retention} days\"})\n else:\n findings.append({\"check\": \"Automatic backups\", \"status\": \"FAIL\", \"detail\": \"No automatic backups — data loss risk on failure\"})\n\n # Security groups from node groups\n sg_ids = set()\n for ng in rg.get(\"NodeGroups\", []):\n for member in ng.get(\"NodeGroupMembers\", []):\n if \"CacheClusterId\" in member:\n try:\n cc = client.describe_cache_clusters(\n CacheClusterId=member[\"CacheClusterId\"],\n ShowCacheNodeInfo=True\n )[\"CacheClusters\"][0]\n for sg in cc.get(\"SecurityGroups\", []):\n sg_ids.add(sg[\"SecurityGroupId\"])\n except (ClientError, BotoCoreError):\n pass\n\n sg_ids_list = list(sg_ids)\n sg_findings = _audit_security_groups(ec2_client, sg_ids_list)\n findings.extend(sg_findings)\n info[\"security_groups\"] = sg_ids_list\n\n # Tags\n try:\n tag_resp = client.list_tags_for_resource(ResourceName=info[\"arn\"])\n tags = {t[\"Key\"]: t[\"Value\"] for t in tag_resp.get(\"TagList\", [])}\n except (ClientError, BotoCoreError):\n tags = {}\n _audit_tags(findings, tags)\n info[\"tags\"] = tags\n\n return {\"info\": info, \"findings\": findings}\n\n\ndef _audit_security_groups(ec2_client, sg_ids):\n \"\"\"Check security groups for overly permissive rules.\"\"\"\n findings = []\n if not sg_ids:\n findings.append({\"check\": \"Security groups\", \"status\": \"WARN\", \"detail\": \"No security groups found\"})\n return findings\n\n try:\n resp = ec2_client.describe_security_groups(GroupIds=sg_ids)\n except Exception as e:\n findings.append({\"check\": \"Security groups\", \"status\": \"ERROR\", \"detail\": f\"Could not inspect: {e}\"})\n return findings\n\n for sg in resp.get(\"SecurityGroups\", []):\n sg_id = sg[\"GroupId\"]\n for rule in sg.get(\"IpPermissions\", []):\n from_port = rule.get(\"FromPort\", 0)\n to_port = rule.get(\"ToPort\", 0)\n for ip_range in rule.get(\"IpRanges\", []):\n cidr = ip_range.get(\"CidrIp\", \"\")\n if cidr == \"0.0.0.0/0\":\n findings.append({\n \"check\": f\"Security group {sg_id}\",\n \"status\": \"WARN\",\n \"detail\": f\"Inbound rule allows 0.0.0.0/0 on ports {from_port}-{to_port} -- no public endpoint exists, but 0.0.0.0/0 violates least-privilege\"\n })\n elif cidr.endswith(\"/0\") or cidr.endswith(\"/8\"):\n findings.append({\n \"check\": f\"Security group {sg_id}\",\n \"status\": \"WARN\",\n \"detail\": f\"Inbound rule allows {cidr} on ports {from_port}-{to_port} -- very broad CIDR\"\n })\n for ip_range in rule.get(\"Ipv6Ranges\", []):\n cidr = ip_range.get(\"CidrIpv6\", \"\")\n if cidr == \"::/0\":\n findings.append({\n \"check\": f\"Security group {sg_id}\",\n \"status\": \"WARN\",\n \"detail\": f\"Inbound rule allows ::/0 (IPv6) on ports {from_port}-{to_port} -- no public endpoint exists, but ::/0 violates least-privilege\"\n })\n\n if not any(f[\"check\"].startswith(\"Security group\") for f in findings):\n findings.append({\"check\": \"Security groups\", \"status\": \"PASS\", \"detail\": f\"No overly permissive rules in {', '.join(sg_ids)}\"})\n\n return findings\n\n\ndef _audit_tags(findings, tags):\n \"\"\"Check for recommended tags.\"\"\"\n recommended = [\"Environment\", \"managed_by\", \"Application\", \"Owner\"]\n missing = [t for t in recommended if t not in tags]\n if not missing:\n findings.append({\"check\": \"Resource tags\", \"status\": \"PASS\", \"detail\": f\"All recommended tags present\"})\n else:\n findings.append({\"check\": \"Resource tags\", \"status\": \"WARN\", \"detail\": f\"Missing recommended tags: {', '.join(missing)}\"})\n\n\ndef format_report(result):\n \"\"\"Format audit results as a human-readable report.\"\"\"\n if \"error\" in result:\n return f\"ERROR: {result['error']}\"\n\n info = result[\"info\"]\n findings = result[\"findings\"]\n\n lines = []\n lines.append(\"=\" * 72)\n lines.append(\"ElastiCache Security Audit Report\")\n lines.append(\"=\" * 72)\n lines.append(\"\")\n\n lines.append(f\" Cache type: {info['cache_type']}\")\n if info[\"cache_type\"] == \"serverless\":\n lines.append(f\" Cache name: {info['cache_name']}\")\n else:\n lines.append(f\" Replication group: {info['replication_group_id']}\")\n lines.append(f\" Node type: {info.get('node_type', 'unknown')}\")\n lines.append(f\" Cluster mode: {'enabled' if info.get('cluster_enabled') else 'disabled'}\")\n lines.append(f\" Node groups: {info.get('num_node_groups', 'unknown')}\")\n lines.append(f\" Engine: {info['engine']} {info.get('engine_version', '')}\")\n lines.append(f\" Status: {info['status']}\")\n lines.append(f\" Auth mode: {info.get('auth_mode', 'unknown')}\")\n lines.append(\"\")\n\n fail_count = sum(1 for f in findings if f[\"status\"] == \"FAIL\")\n warn_count = sum(1 for f in findings if f[\"status\"] == \"WARN\")\n pass_count = sum(1 for f in findings if f[\"status\"] == \"PASS\")\n\n lines.append(f\" Summary: {pass_count} passed, {warn_count} warnings, {fail_count} failures\")\n lines.append(\"\")\n\n lines.append(\"-\" * 72)\n\n status_order = {\"FAIL\": 0, \"WARN\": 1, \"ERROR\": 2, \"INFO\": 3, \"PASS\": 4}\n sorted_findings = sorted(findings, key=lambda f: status_order.get(f[\"status\"], 5))\n\n for f in sorted_findings:\n icon = {\"PASS\": \"[OK] \", \"FAIL\": \"[FAIL]\", \"WARN\": \"[WARN]\", \"INFO\": \"[INFO]\", \"ERROR\": \"[ERR] \"}\n lines.append(f\" {icon.get(f['status'], '[?] ')} {f['check']}\")\n lines.append(f\" {f['detail']}\")\n lines.append(\"\")\n\n if fail_count > 0:\n lines.append(\"=\" * 72)\n lines.append(\"RECOMMENDED ACTIONS\")\n lines.append(\"=\" * 72)\n lines.append(\"\")\n for f in sorted_findings:\n if f[\"status\"] == \"FAIL\":\n check = f[\"check\"]\n if \"TLS\" in check:\n lines.append(\" - Enable in-transit encryption. For existing node-based clusters, use a\")\n lines.append(\" two-step process: first set transit-encryption-mode to 'preferred',\")\n lines.append(\" then after migrating all clients to TLS, set it to 'required'.\")\n elif \"Encryption at rest\" in check:\n lines.append(\" - Enable at-rest encryption. For existing clusters, this requires\")\n lines.append(\" creating a new cluster with AtRestEncryptionEnabled=true and migrating.\")\n elif \"Authentication\" in check:\n lines.append(\" - Configure RBAC: create users, a user group, and attach it to the cache.\")\n lines.append(\" See references/setup/iam-policies.md for IAM auth patterns.\")\n elif \"backups\" in check.lower():\n lines.append(\" - Enable automatic backups: modify the replication group with\")\n lines.append(\" SnapshotRetentionLimit > 0 (recommended: 7 days).\")\n elif \"0.0.0.0/0\" in f[\"detail\"]:\n lines.append(f\" - Restrict {check}: remove the 0.0.0.0/0 inbound rule and scope to\")\n lines.append(\" application security groups only.\")\n lines.append(\"\")\n\n lines.append(\" Note: This is an informational review, not a security certification.\")\n lines.append(\" Consult your security team before production deployment.\")\n lines.append(\"\")\n return \"\\n\".join(lines)\n\n\nif __name__ == \"__main__\":\n parser = argparse.ArgumentParser(description=\"ElastiCache Security Audit\")\n group = parser.add_mutually_exclusive_group(required=True)\n group.add_argument(\"--serverless\", metavar=\"CACHE_NAME\", help=\"Audit a serverless cache\")\n group.add_argument(\"--replication-group\", metavar=\"RG_ID\", help=\"Audit a node-based replication group\")\n parser.add_argument(\"--region\", default=\"us-east-1\", help=\"AWS region\")\n parser.add_argument(\"--profile\", default=None, help=\"AWS profile\")\n parser.add_argument(\"--json\", action=\"store_true\", help=\"Output as JSON\")\n args = parser.parse_args()\n\n try:\n session = boto3.Session(profile_name=args.profile, region_name=args.region)\n ec_client = session.client(\"elasticache\")\n ec2_client = session.client(\"ec2\")\n except Exception as e:\n print(f\"ERROR: Failed to create AWS session: {e}\", file=sys.stderr)\n sys.exit(1)\n\n try:\n if args.serverless:\n result = audit_serverless(ec_client, ec2_client, args.serverless)\n else:\n result = audit_replication_group(ec_client, ec2_client, args.replication_group)\n except Exception as e:\n error_code = \"\"\n if hasattr(e, \"response\"):\n error_code = e.response.get(\"Error\", {}).get(\"Code\", \"\")\n if error_code in (\"AccessDeniedException\", \"AccessDenied\"):\n print(f\"ERROR: Access denied. Ensure your IAM role/user has elasticache:Describe*, \"\n f\"elasticache:ListTagsForResource, and ec2:DescribeSecurityGroups permissions.\")\n else:\n print(f\"ERROR: {e}\")\n sys.exit(1)\n\n if args.json:\n print(json.dumps(result, indent=2, default=str))\n else:\n print(format_report(result))\n\n if \"error\" in result:\n sys.exit(1)\n fail_count = sum(1 for f in result.get(\"findings\", []) if f[\"status\"] == \"FAIL\")\n sys.exit(1 if fail_count > 0 else 0)\n","content_type":"text/x-python; charset=utf-8","language":"python","size":19998,"content_sha256":"2f2b42ba13eceb7b70a9b7c5f3c70fa12ea8acc98b483ce0653a8f6e29b221c5"},{"filename":"scripts/serverless_estimator.py","content":"#!/usr/bin/env python3\n\"\"\"\nElastiCache Serverless Cost Estimator\n\nProvides approximate cost estimates for what ElastiCache provisioned\nclusters would cost on ElastiCache Serverless, or for sizing a new\nserverless cache. These are estimates only. Actual serverless costs\ndepend on real-time workload characteristics and may differ from the\nvalues produced by this tool. For accurate billing data, use CloudWatch\nElastiCacheProcessingUnits and BytesUsedForCache metrics on a running\nserverless cache.\n\nTwo estimation modes:\n Simple: Provide avg memory + daily command count per cluster.\n Each command is estimated at MAX(1, avg_payload_kb) ECPUs.\n This is a quick approximation for initial cost comparison.\n\n Detailed: Also provide per-command stats (from INFO commandstats).\n Classifies commands as simple (1 ECPU) or complex\n (estimated as MAX(calls, usec/3) ECPUs based on the\n assumption that 1 ECPU corresponds to approximately\n 3 microseconds of vCPU time, derived from the AWS pricing\n blog example). Takes the higher of the CPU and network\n components. More accurate than simple mode but still an\n approximation.\n\nIMPORTANT: The ECPU formulas used in this tool are heuristic approximations\nbased on publicly available AWS documentation and pricing examples. AWS does\nnot publish exact ECPU calculation formulas. The 3-microsecond-per-ECPU\nassumption for complex commands is derived from the AWS pricing blog and may\nnot apply uniformly to all command types. See command_classifier.py for\ndetails on command classification.\n\nPricing is fetched live from the AWS Bulk Pricing API and reflects current\npublished rates. See https://aws.amazon.com/elasticache/pricing/\n\nNote: INFO commandstats returns cumulative counts that keep incrementing\nfrom server start. If you include an \"uptime_days\" column in the\ncommandstats CSV, the estimator divides calls and usec by that value to\nnormalize to a daily rate. The --endpoint mode handles this automatically\nby taking two snapshots and computing the delta.\n\nUsage:\n python3 serverless_estimator.py --input clusters.csv\n python3 serverless_estimator.py --input clusters.csv --commandstats stats.csv\n python3 serverless_estimator.py --endpoint my-cluster.abc123.use1.cache.amazonaws.com:6379\n python3 serverless_estimator.py --endpoint my-cluster.abc123.use1.cache.amazonaws.com --instance-type cache.r7g.large\n python3 serverless_estimator.py --input clusters.csv --pricing pricing.csv\n python3 serverless_estimator.py --input clusters.csv --output estimate.csv --json\n\"\"\"\nimport argparse\nimport csv\nimport json\nimport math\nimport os\nimport re\nimport sys\nfrom typing import Dict, List, Optional\n\nsys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))\n\nfrom command_classifier import estimate_ecpus_from_commandstats\nfrom pricing import PricingLoader\n\nHOURS_PER_MONTH = 730\n\nMIN_STORAGE_GB = {\n \"valkey\": 0.1, # 100 MB\n \"redis\": 1.0, # 1 GB\n \"memcached\": 1.0, # 1 GB\n}\n\n# Number of sample points for sinusoidal workload modeling.\n# 24 = one sample per hour over a daily cycle.\n_SIN_SAMPLES = 24\n\n\ndef _sine_wave_samples(min_val, max_val, n=_SIN_SAMPLES):\n \"\"\"Generate n hourly samples of a sine wave oscillating between min and max.\n\n Models a daily traffic pattern:\n value(t) = min + (max - min) * (1 + sin(2*pi*t/n)) / 2\n\n Returns a list of n float values representing the workload at each hour.\n \"\"\"\n amplitude = (max_val - min_val) / 2.0\n midpoint = (max_val + min_val) / 2.0\n return [midpoint + amplitude * math.sin(2 * math.pi * i / n) for i in range(n)]\n\n\ndef compute_burst_multiplier(avg_val, min_val=None, max_val=None, peak_to_avg=None):\n \"\"\"Compute a cost multiplier for bursty (sinusoidal) workloads.\n\n Serverless bills per-hour. A workload that swings between min and peak\n follows a sine curve over the day. This function samples the curve at\n 24 hourly points and returns the ratio of the sine-wave average to the\n customer-provided average.\n\n Why this matters: if the customer says \"my average is X\" but their actual\n traffic swings from min to peak, the true average of the sine curve\n ((min+max)/2) may differ from X. The multiplier corrects for that.\n\n The caller can specify the burst in two ways:\n 1. min_val + max_val: explicit floor and ceiling of the daily cycle\n 2. peak_to_avg: ratio of peak to average (e.g., 3.0 means peak is 3x avg)\n\n Returns 1.0 (no adjustment) if no burst parameters are provided or if\n the workload is flat.\n \"\"\"\n if peak_to_avg is not None and peak_to_avg > 1.0:\n max_val = avg_val * peak_to_avg\n min_val = max(0, avg_val * (2 - peak_to_avg))\n\n if min_val is None or max_val is None:\n return 1.0\n if max_val \u003c= min_val or max_val \u003c= 0:\n return 1.0\n if avg_val \u003c= 0:\n return 1.0\n\n samples = _sine_wave_samples(min_val, max_val)\n sine_avg = sum(samples) / len(samples)\n\n return sine_avg / avg_val\n\n\ndef load_clusters(path: str) -> List[Dict]:\n \"\"\"Load cluster data from CSV.\n\n Required columns:\n cluster_name, instance_type, region, engine, node_count,\n avg_memory_gb, daily_commands\n\n Optional columns:\n primary_nodes -- defaults to max(1, node_count // 2). This is a\n rough heuristic; for CMD clusters (1 primary +\n N replicas) or CME clusters (1 primary per shard),\n provide the actual value via --primary-nodes or\n the primary_nodes CSV column for accurate results.\n current_monthly_cost -- if known; otherwise computed from list price\n avg_payload_kb -- average payload size in KB (default 1.0).\n ECPUs scale linearly with payload: 3.2 KB = 3.2 ECPUs per op.\n peak_commands -- peak daily commands (for bursty workload modeling)\n min_commands -- minimum daily commands (for bursty workload modeling)\n peak_to_avg_ratio -- ratio of peak to average commands (alternative to min/max)\n peak_memory_gb -- peak memory in GB (for bursty storage modeling)\n min_memory_gb -- minimum memory in GB (for bursty storage modeling)\n \"\"\"\n clusters = []\n with open(path, newline=\"\") as f:\n reader = csv.DictReader(f)\n for row in reader:\n node_count = int(row.get(\"node_count\", 1))\n primary_nodes = int(row.get(\"primary_nodes\", 0))\n if primary_nodes == 0:\n # Heuristic: assumes half the nodes are primaries. This is only\n # accurate for 1-replica setups; actual primary count depends on\n # cluster topology (CMD: 1 primary + N replicas, CME: 1 primary\n # per shard). Provide the primary_nodes column for accuracy.\n primary_nodes = max(1, node_count // 2)\n\n clusters.append({\n \"cluster_name\": row[\"cluster_name\"].strip(),\n \"instance_type\": row.get(\"instance_type\", \"\").strip(),\n \"region\": row.get(\"region\", \"us-east-1\").strip(),\n \"engine\": row.get(\"engine\", \"valkey\").strip().lower(),\n \"node_count\": node_count,\n \"primary_nodes\": primary_nodes,\n \"avg_memory_gb\": float(row.get(\"avg_memory_gb\", 0)),\n \"daily_commands\": float(row.get(\"daily_commands\", 0)),\n \"current_monthly_cost\": float(row.get(\"current_monthly_cost\", 0) or 0),\n \"avg_payload_kb\": float(row.get(\"avg_payload_kb\", 1.0) or 1.0),\n \"peak_commands\": float(row.get(\"peak_commands\", 0) or 0),\n \"min_commands\": float(row.get(\"min_commands\", 0) or 0),\n \"peak_to_avg_ratio\": float(row.get(\"peak_to_avg_ratio\", 0) or 0),\n \"peak_memory_gb\": float(row.get(\"peak_memory_gb\", 0) or 0),\n \"min_memory_gb\": float(row.get(\"min_memory_gb\", 0) or 0),\n })\n return clusters\n\n\ndef parse_commandstats_file(path: str) -> Dict[str, dict]:\n \"\"\"Parse a commandstats CSV.\n\n Expected columns: cluster_name, command, calls, usec\n\n Optional columns:\n uptime_days - days since server restart. When present, calls and\n usec are divided by this value to normalize cumulative\n INFO commandstats output to a daily rate. Without it\n the raw cumulative values are used as-is.\n\n Returns a dict keyed by cluster_name. Each value is a dict of\n {command: {\"calls\": int, \"usec\": int}}. A special key\n \"_normalized\" (bool) indicates whether uptime_days normalization\n was applied for that cluster.\n \"\"\"\n result = {}\n with open(path, newline=\"\") as f:\n for row in csv.DictReader(f):\n cluster = row[\"cluster_name\"].strip()\n cmd = row[\"command\"].strip().lower()\n calls = int(row.get(\"calls\", 0))\n usec = int(row.get(\"usec\", 0))\n\n uptime_raw = row.get(\"uptime_days\", \"\").strip()\n uptime_days = float(uptime_raw) if uptime_raw else None\n\n if uptime_days and uptime_days > 0:\n calls = int(calls / uptime_days)\n usec = int(usec / uptime_days)\n normalized = True\n else:\n normalized = False\n\n if cluster not in result:\n result[cluster] = {\"_normalized\": normalized} # type: ignore[dict-item]\n result[cluster][cmd] = {\"calls\": calls, \"usec\": usec} # type: ignore[assignment]\n # If any row for this cluster has uptime_days, mark as normalized\n if normalized:\n result[cluster][\"_normalized\"] = True\n return result\n\n\ndef parse_commandstats_info(text: str) -> dict:\n \"\"\"Parse raw INFO commandstats output from valkey-cli (or redis-cli).\n\n Example line: cmdstat_get:calls=1000,usec=1500,usec_per_call=1.50,rejected_calls=0,failed_calls=0\n \"\"\"\n result = {}\n for line in text.strip().splitlines():\n line = line.strip()\n if not line or line.startswith(\"#\"):\n continue\n match = re.match(r\"cmdstat_(\\w+):calls=(\\d+),usec=(\\d+)\", line)\n if match:\n cmd = match.group(1).lower()\n result[cmd] = {\n \"calls\": int(match.group(2)),\n \"usec\": int(match.group(3)),\n }\n return result\n\n\ndef estimate_cluster(\n cluster: Dict,\n pricing: PricingLoader,\n commandstats: Optional[dict] = None,\n) -> Dict:\n \"\"\"Estimate serverless cost for one cluster.\"\"\"\n region = cluster[\"region\"]\n engine = cluster[\"engine\"]\n node_count = cluster[\"node_count\"]\n instance_type = cluster[\"instance_type\"]\n\n # --- Current provisioned cost ---\n if cluster[\"current_monthly_cost\"] > 0:\n current_cost = cluster[\"current_monthly_cost\"]\n cost_source = \"provided\"\n elif instance_type:\n try:\n hourly = pricing.get_node_hourly_rate(region, instance_type, engine)\n current_cost = hourly * HOURS_PER_MONTH * node_count\n cost_source = \"list_price\"\n except ValueError:\n current_cost = 0\n cost_source = \"unknown\"\n else:\n current_cost = 0\n cost_source = \"unknown\"\n\n # --- Serverless storage ---\n avg_memory_gb = cluster[\"avg_memory_gb\"]\n min_gb = MIN_STORAGE_GB.get(engine, 1.0)\n billing_memory_gb = max(avg_memory_gb, min_gb)\n used_minimum = avg_memory_gb \u003c min_gb\n\n # Bursty storage: if peak/min memory provided, adjust\n storage_burst = compute_burst_multiplier(\n avg_memory_gb,\n min_val=cluster.get(\"min_memory_gb\") or None,\n max_val=cluster.get(\"peak_memory_gb\") or None,\n )\n adjusted_memory_gb = billing_memory_gb * storage_burst\n\n monthly_gb_hours = adjusted_memory_gb * HOURS_PER_MONTH\n storage_rate = pricing.get_serverless_storage_rate(region, engine)\n storage_cost = monthly_gb_hours * storage_rate\n\n # --- Serverless ECPUs ---\n ecpu_rate = pricing.get_serverless_ecpu_rate(region, engine)\n\n # Bursty commands: if peak/min commands or peak_to_avg_ratio provided, adjust\n ecpu_burst = compute_burst_multiplier(\n cluster[\"daily_commands\"],\n min_val=cluster.get(\"min_commands\") or None,\n max_val=cluster.get(\"peak_commands\") or None,\n peak_to_avg=cluster.get(\"peak_to_avg_ratio\") or None,\n )\n\n if commandstats and engine == \"memcached\":\n print(\"WARNING: commandstats mode uses Redis/Valkey INFO format, which is \"\n \"incompatible with Memcached. ECPU estimates for Memcached cluster '{}' \"\n \"will be inaccurate. Falling back to simple estimation.\".format(\n cluster[\"cluster_name\"]),\n file=sys.stderr)\n commandstats = None\n\n if commandstats:\n # _normalized is a metadata flag, not a command; exclude before estimation\n cs_normalized = commandstats.get(\"_normalized\", False)\n cs_data = {k: v for k, v in commandstats.items() if k != \"_normalized\"}\n ecpu_result = estimate_ecpus_from_commandstats(cs_data)\n daily_cpu_ecpus = ecpu_result[\"total_ecpus\"]\n # Network component: each command costs MAX(1, avg_payload_kb) ECPUs.\n # This approximates the \"1 ECPU per KB transferred\" pricing rule.\n # See: https://aws.amazon.com/elasticache/pricing/\n avg_payload_kb = cluster.get(\"avg_payload_kb\", 1.0) or 1.0\n ecpu_per_request = max(1.0, avg_payload_kb)\n daily_net_ecpus = cluster[\"daily_commands\"] * ecpu_per_request\n # Serverless charges the higher of vCPU time or data transferred.\n # See: https://aws.amazon.com/blogs/database/unlock-on-demand-cost-optimized-performance-with-amazon-elasticache-serverless/\n daily_ecpus = max(daily_cpu_ecpus, daily_net_ecpus)\n monthly_ecpus = daily_ecpus * 30 * ecpu_burst\n ecpu_mode = \"detailed\"\n fixed_ecpus = ecpu_result[\"fixed_ecpus\"] * 30\n nonfixed_ecpus = ecpu_result[\"nonfixed_ecpus\"] * 30\n internal_calls = ecpu_result[\"internal_calls\"] * 30\n else:\n avg_payload_kb = cluster.get(\"avg_payload_kb\", 1.0) or 1.0\n ecpu_per_request = max(1.0, avg_payload_kb)\n monthly_ecpus = cluster[\"daily_commands\"] * 30 * ecpu_per_request * ecpu_burst\n ecpu_mode = \"simple\"\n fixed_ecpus = None\n nonfixed_ecpus = None\n internal_calls = None\n\n ecpu_cost = monthly_ecpus * ecpu_rate\n total_serverless = storage_cost + ecpu_cost\n\n # --- Comparison ---\n diff = total_serverless - current_cost if current_cost > 0 else None\n diff_pct = (diff / current_cost * 100) if diff is not None and current_cost > 0 else None\n\n # --- Notes ---\n notes = []\n if used_minimum:\n notes.append(\"Min storage applied ({} GB)\".format(min_gb))\n if cost_source == \"list_price\":\n notes.append(\"Current cost: on-demand list price\")\n if cost_source == \"unknown\":\n notes.append(\"Current cost unknown - provide current_monthly_cost or instance_type\")\n if ecpu_mode == \"simple\":\n if avg_payload_kb > 1:\n notes.append(\"ECPU: {} ECPUs per request ({} KB avg payload)\".format(\n round(ecpu_per_request, 1), round(avg_payload_kb, 1)))\n else:\n notes.append(\"ECPU: 1 cmd = 1 ECPU (provide avg_payload_kb for better accuracy)\")\n if commandstats and not cs_normalized:\n notes.append(\"Commandstats not normalized by uptime; provide uptime_days column for accuracy\")\n if ecpu_burst > 1.0:\n notes.append(\"Burst adjustment applied: {:.2f}x (peak/avg commands)\".format(ecpu_burst))\n if storage_burst > 1.0:\n notes.append(\"Burst adjustment applied: {:.2f}x (peak/avg memory)\".format(storage_burst))\n\n # Serverless compatibility warnings\n if engine in (\"redis\", \"valkey\"):\n notes.append(\"Serverless requires cluster-mode-enabled clients and TLS; Global Data Store and Data Tiering are not supported\")\n\n return {\n \"cluster_name\": cluster[\"cluster_name\"],\n \"region\": region,\n \"engine\": engine,\n \"instance_type\": instance_type,\n \"node_count\": node_count,\n \"primary_nodes\": cluster[\"primary_nodes\"],\n \"current_monthly_cost\": round(current_cost, 2),\n \"cost_source\": cost_source,\n \"avg_memory_gb\": round(avg_memory_gb, 4),\n \"billing_memory_gb\": round(billing_memory_gb, 4),\n \"monthly_gb_hours\": round(monthly_gb_hours, 2),\n \"storage_rate_per_gb_hour\": storage_rate,\n \"storage_cost\": round(storage_cost, 2),\n \"monthly_ecpus\": round(monthly_ecpus),\n \"ecpu_rate_per_million\": round(ecpu_rate * 1_000_000, 4),\n \"ecpu_cost\": round(ecpu_cost, 2),\n \"ecpu_mode\": ecpu_mode,\n \"fixed_ecpus\": round(fixed_ecpus) if fixed_ecpus is not None else None,\n \"nonfixed_ecpus\": round(nonfixed_ecpus) if nonfixed_ecpus is not None else None,\n \"internal_calls_excluded\": round(internal_calls) if internal_calls is not None else None,\n \"serverless_total\": round(total_serverless, 2),\n \"diff\": round(diff, 2) if diff is not None else None,\n \"diff_pct\": round(diff_pct, 1) if diff_pct is not None else None,\n \"notes\": \"; \".join(notes),\n }\n\n\ndef write_csv(results: List[Dict], path: str):\n \"\"\"Write results to CSV.\"\"\"\n if not results:\n return\n fields = [\n \"cluster_name\", \"region\", \"engine\", \"instance_type\",\n \"node_count\", \"primary_nodes\",\n \"current_monthly_cost\", \"cost_source\",\n \"avg_memory_gb\", \"billing_memory_gb\",\n \"monthly_gb_hours\", \"storage_cost\",\n \"monthly_ecpus\", \"ecpu_cost\", \"ecpu_mode\",\n \"serverless_total\", \"diff\", \"diff_pct\", \"notes\",\n ]\n with open(path, \"w\", newline=\"\") as f:\n writer = csv.DictWriter(f, fieldnames=fields, extrasaction=\"ignore\")\n writer.writeheader()\n writer.writerows(results)\n print(\"Output written to: {}\".format(path))\n\n\ndef print_summary(results: List[Dict]):\n \"\"\"Print summary to console.\"\"\"\n total_current = sum(r[\"current_monthly_cost\"] for r in results)\n total_sl = sum(r[\"serverless_total\"] for r in results)\n total_stor = sum(r[\"storage_cost\"] for r in results)\n total_ecpu = sum(r[\"ecpu_cost\"] for r in results)\n cheaper = sum(1 for r in results if r[\"diff\"] is not None and r[\"diff\"] \u003c 0)\n more_exp = sum(1 for r in results if r[\"diff\"] is not None and r[\"diff\"] > 0)\n\n print()\n print(\"=\" * 65)\n print(\" ElastiCache Serverless Cost Estimate\")\n print(\"=\" * 65)\n print(\" Clusters analyzed: {}\".format(len(results)))\n print(\" Total nodes: {}\".format(sum(r[\"node_count\"] for r in results)))\n print()\n print(\" Current provisioned cost: ${:>12,.2f} /month\".format(total_current))\n print(\" Estimated serverless: ${:>12,.2f} /month\".format(total_sl))\n print(\" Data storage: ${:>12,.2f}\".format(total_stor))\n print(\" ECPUs: ${:>12,.2f}\".format(total_ecpu))\n print()\n if total_current > 0:\n savings = total_current - total_sl\n pct = savings / total_current * 100\n print(\" Estimated savings: ${:>12,.2f} ({:.1f}%)\".format(savings, pct))\n print(\" Clusters cheaper on SL: {}\".format(cheaper))\n print(\" Clusters more expensive: {}\".format(more_exp))\n print(\"=\" * 65)\n print()\n print(\" {:\u003c30s} {:>10s} {:>10s} {:>8s}\".format(\n \"Cluster\", \"Current\", \"Serverless\", \"Change\"))\n print(\" {} {} {} {}\".format(\"-\" * 30, \"-\" * 10, \"-\" * 10, \"-\" * 8))\n for r in sorted(results, key=lambda x: x.get(\"diff\") or 0):\n name = r[\"cluster_name\"][:30]\n cur = \"${:,.0f}\".format(r[\"current_monthly_cost\"])\n sl = \"${:,.0f}\".format(r[\"serverless_total\"])\n chg = \"{:+.0f}%\".format(r[\"diff_pct\"]) if r[\"diff_pct\"] is not None else \"N/A\"\n print(\" {:\u003c30s} {:>10s} {:>10s} {:>8s}\".format(name, cur, sl, chg))\n print()\n print(\" ECPU estimates assume 1 KB avg payload unless avg_payload_kb is provided.\")\n print(\" ECPUs scale linearly with payload size (e.g. 3 KB = 3x ECPUs).\")\n print()\n print(\" IMPORTANT: These are approximate estimates. Actual serverless costs depend\")\n print(\" on real-time workload characteristics and may differ. For accurate billing,\")\n print(\" use CloudWatch metrics on a running serverless cache.\")\n print(\" See https://aws.amazon.com/elasticache/pricing/\")\n print()\n\n\ndef collect_from_endpoint(endpoint, cluster_name=None, instance_type=\"\",\n region=\"us-east-1\", engine=\"valkey\", node_count=2,\n avg_payload_kb=1.0, use_tls=True, sample_seconds=60):\n \"\"\"Connect to a live Valkey/Redis endpoint and collect metrics.\n\n Takes two INFO snapshots separated by sample_seconds and computes\n deltas. This gives the actual traffic rate during the sample window,\n which is more accurate than dividing cumulative totals by uptime.\n\n Returns (cluster_dict, commandstats_dict) ready for estimate_cluster().\n \"\"\"\n try:\n import valkey as client_lib\n except ImportError:\n print(\"Error: the 'valkey' package is required for --endpoint mode.\")\n print(\"Install with: pip install valkey\")\n sys.exit(1)\n\n import time\n\n # Parse host:port\n if \":\" in endpoint and not endpoint.startswith(\"[\"):\n host, port_str = endpoint.rsplit(\":\", 1)\n port = int(port_str)\n else:\n host = endpoint\n port = 6379\n\n if not cluster_name:\n cluster_name = host.split(\".\")[0]\n\n print(\"Connecting to {}:{}{}...\".format(host, port, \" (TLS)\" if use_tls else \"\"))\n r = client_lib.Redis(host=host, port=port, ssl=use_tls,\n ssl_cert_reqs=None, decode_responses=True,\n socket_connect_timeout=10)\n print(\" PING: {}\".format(r.ping()))\n\n # Collect memory and replication (point-in-time, no delta needed)\n mem_info = r.info(\"memory\")\n repl_info = r.info(\"replication\")\n dataset_bytes = mem_info.get(\"used_memory_dataset\", mem_info.get(\"used_memory\", 0))\n dataset_gb = dataset_bytes / (1024 ** 3)\n role = repl_info.get(\"role\", \"unknown\")\n\n print(\" Role: {}\".format(role))\n print(\" Memory: {:.4f} GB ({:,.0f} bytes)\".format(dataset_gb, dataset_bytes))\n\n # Snapshot 1\n print(\" Taking snapshot 1...\")\n cs1 = r.info(\"commandstats\")\n stats1 = r.info(\"stats\")\n t1 = time.time()\n\n # Wait\n print(\" Sampling for {} seconds...\".format(sample_seconds))\n time.sleep(sample_seconds)\n\n # Snapshot 2\n print(\" Taking snapshot 2...\")\n cs2 = r.info(\"commandstats\")\n stats2 = r.info(\"stats\")\n t2 = time.time()\n\n elapsed = t2 - t1\n elapsed_days = elapsed / 86400\n\n # Compute commandstats deltas\n commandstats = {}\n total_delta_calls = 0\n for key in cs2:\n cmd = key.replace(\"cmdstat_\", \"\")\n calls_delta = cs2[key].get(\"calls\", 0) - cs1.get(key, {}).get(\"calls\", 0)\n usec_delta = cs2[key].get(\"usec\", 0) - cs1.get(key, {}).get(\"usec\", 0)\n if calls_delta > 0:\n # Scale to daily rate\n daily_calls = int(calls_delta / elapsed_days)\n daily_usec = int(usec_delta / elapsed_days)\n commandstats[cmd] = {\"calls\": daily_calls, \"usec\": daily_usec}\n total_delta_calls += calls_delta\n\n commandstats[\"_normalized\"] = True # type: ignore[assignment]\n daily_commands = total_delta_calls / elapsed_days if elapsed_days > 0 else 0\n\n # Also get total commands delta for simple mode\n total_cmds_delta = stats2.get(\"total_commands_processed\", 0) - stats1.get(\"total_commands_processed\", 0)\n daily_commands_total = total_cmds_delta / elapsed_days if elapsed_days > 0 else 0\n\n print(\" Sample window: {:.0f} seconds\".format(elapsed))\n print(\" Commands in window: {:,}\".format(total_cmds_delta))\n print(\" Daily rate: {:,.0f} commands/day\".format(daily_commands_total))\n\n top_cmds = sorted(commandstats.items(),\n key=lambda x: x[1].get(\"calls\", 0) if isinstance(x[1], dict) else 0,\n reverse=True)[:5]\n print(\" Top commands: {}\".format(\n \", \".join(\"{} ({:,}/day)\".format(k, v[\"calls\"])\n for k, v in top_cmds if isinstance(v, dict))))\n\n cluster = {\n \"cluster_name\": cluster_name,\n \"instance_type\": instance_type,\n \"region\": region,\n \"engine\": engine.lower(),\n \"node_count\": node_count,\n \"primary_nodes\": max(1, node_count // 2),\n \"avg_memory_gb\": dataset_gb,\n \"daily_commands\": daily_commands_total,\n \"current_monthly_cost\": 0,\n \"avg_payload_kb\": avg_payload_kb,\n \"peak_commands\": 0,\n \"min_commands\": 0,\n \"peak_to_avg_ratio\": 0,\n \"peak_memory_gb\": 0,\n \"min_memory_gb\": 0,\n }\n\n return cluster, commandstats\n\n\ndef main():\n parser = argparse.ArgumentParser(\n description=\"Estimate ElastiCache Serverless costs from provisioned cluster metrics\"\n )\n parser.add_argument(\"--input\", \"-i\",\n help=\"CSV with cluster data (see README for format)\")\n parser.add_argument(\"--endpoint\", \"-e\",\n help=\"Connect directly to a Valkey/Redis endpoint (host:port or host)\")\n parser.add_argument(\"--cluster-name\", default=None,\n help=\"Cluster name (used with --endpoint, default: derived from host)\")\n parser.add_argument(\"--instance-type\", default=\"\",\n help=\"Instance type (used with --endpoint, e.g., cache.r7g.large)\")\n parser.add_argument(\"--region\", default=\"us-east-1\",\n help=\"AWS region (used with --endpoint, default: us-east-1)\")\n parser.add_argument(\"--engine\", default=\"valkey\",\n help=\"Engine (used with --endpoint, default: valkey)\")\n parser.add_argument(\"--node-count\", type=int, default=2,\n help=\"Node count (used with --endpoint, default: 2)\")\n parser.add_argument(\"--avg-payload-kb\", type=float, default=1.0,\n help=\"Average payload size in KB (default: 1.0)\")\n parser.add_argument(\"--no-tls\", action=\"store_true\",\n help=\"Disable TLS when connecting via --endpoint\")\n parser.add_argument(\"--sample-seconds\", type=int, default=60,\n help=\"Seconds to sample when using --endpoint (default: 60). \"\n \"Takes two snapshots this far apart and computes deltas.\")\n parser.add_argument(\"--commandstats\", \"-c\",\n help=\"CSV with per-command stats for detailed ECPU estimation\")\n parser.add_argument(\"--pricing\", \"-p\",\n help=\"Pricing CSV (optional - fetches live from AWS if omitted)\")\n parser.add_argument(\"--output\", \"-o\", default=\"serverless_estimate.csv\",\n help=\"Output CSV path (default: serverless_estimate.csv)\")\n parser.add_argument(\"--json\", action=\"store_true\",\n help=\"Also write JSON output\")\n args = parser.parse_args()\n\n if not args.input and not args.endpoint:\n parser.error(\"Either --input (CSV) or --endpoint (host:port) is required\")\n\n pricing = PricingLoader(args.pricing)\n\n if args.endpoint:\n # Direct endpoint mode: connect, collect, normalize, estimate\n cluster, commandstats = collect_from_endpoint(\n args.endpoint,\n cluster_name=args.cluster_name,\n instance_type=args.instance_type,\n region=args.region,\n engine=args.engine,\n node_count=args.node_count,\n avg_payload_kb=args.avg_payload_kb,\n use_tls=not args.no_tls,\n sample_seconds=args.sample_seconds,\n )\n results = [estimate_cluster(cluster, pricing, commandstats)]\n else:\n print(\"Loading clusters from: {}\".format(args.input))\n clusters = load_clusters(args.input)\n print(\" {} clusters loaded\".format(len(clusters)))\n\n all_commandstats = None\n if args.commandstats:\n print(\"Loading commandstats from: {}\".format(args.commandstats))\n all_commandstats = parse_commandstats_file(args.commandstats)\n print(\" Stats for {} clusters\".format(len(all_commandstats)))\n\n results = []\n for cluster in clusters:\n cs = all_commandstats.get(cluster[\"cluster_name\"]) if all_commandstats else None\n results.append(estimate_cluster(cluster, pricing, cs))\n\n write_csv(results, args.output)\n print_summary(results)\n\n if args.json:\n base, _ = os.path.splitext(args.output)\n json_path = base + \".json\"\n with open(json_path, \"w\") as f:\n json.dump(results, f, indent=2, default=str)\n print(\"JSON written to: {}\".format(json_path))\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":29076,"content_sha256":"4dc57fd99a316ab4f6aa5161d9ea562e3a0cf11ac7aba6a06583b1a8463ee49b"},{"filename":"scripts/start_tunnel.py","content":"#!/usr/bin/env python3\n\"\"\"\nStart an SSM port-forwarding tunnel to an ElastiCache endpoint.\n\nWraps the AWS SSM StartPortForwardingSessionToRemoteHost document into a\nsingle command with pre-flight checks. Designed to be invoked by an AI agent\nor a human developer, with clear error messages at every failure point.\n\nDependencies:\n pip install boto3\n\nPrerequisites:\n - AWS CLI v2 installed\n - Session Manager plugin installed\n (https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-working-with-install-plugin.html)\n - An SSM-managed EC2 instance in the same VPC as the cache\n (use find_tunnel_host.py to locate one)\n\nUsage:\n python start_tunnel.py \\\\\n --instance-id i-0abc123def456 \\\\\n --cache-host my-cache.serverless.use1.cache.amazonaws.com\n\n python start_tunnel.py \\\\\n --instance-id i-0abc123def456 \\\\\n --cache-host my-cache.serverless.use1.cache.amazonaws.com \\\\\n --cache-port 6379 --local-port 6379 \\\\\n --region us-west-2 --profile my-profile\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport json\nimport os\nimport select\nimport shutil\nimport signal\nimport socket\nimport subprocess\nimport sys\nimport time\n\ntry:\n import boto3\nexcept ImportError:\n print(\n \"Error: the 'boto3' package is required.\\n\"\n \"Install it with:\\n\"\n \" pip install boto3\"\n )\n sys.exit(2)\n\n\ndef check_ssm_plugin() -> bool:\n \"\"\"Verify the Session Manager plugin is installed.\"\"\"\n return shutil.which(\"session-manager-plugin\") is not None\n\n\ndef check_aws_cli() -> bool:\n \"\"\"Verify the AWS CLI is installed.\"\"\"\n return shutil.which(\"aws\") is not None\n\n\ndef check_instance_ssm_status(instance_id: str, session: boto3.Session) -> bool:\n \"\"\"Verify the target instance is online in SSM.\"\"\"\n ssm = session.client(\"ssm\")\n try:\n resp = ssm.describe_instance_information(\n Filters=[{\"Key\": \"InstanceIds\", \"Values\": [instance_id]}]\n )\n instances = resp.get(\"InstanceInformationList\", [])\n if not instances:\n return False\n return instances[0].get(\"PingStatus\") == \"Online\"\n except Exception as exc:\n print(f\"Warning: could not verify SSM status: {exc}\")\n return False\n\n\ndef check_instance_running(instance_id: str, session: boto3.Session) -> bool:\n \"\"\"Verify the target EC2 instance is in a running state.\"\"\"\n ec2 = session.client(\"ec2\")\n try:\n resp = ec2.describe_instance_status(\n InstanceIds=[instance_id], IncludeAllInstances=True\n )\n statuses = resp.get(\"InstanceStatuses\", [])\n if not statuses:\n return False\n return statuses[0][\"InstanceState\"][\"Name\"] == \"running\"\n except Exception as exc:\n print(f\"Warning: could not verify instance state: {exc}\")\n return False\n\n\ndef validate_port(port: int, label: str = \"port\") -> None:\n \"\"\"Validate that a port number is in the usable range (1-65535).\"\"\"\n if not isinstance(port, int) or port \u003c 1 or port > 65535:\n print(f\" [FAIL] {label} must be between 1 and 65535 (got {port})\")\n sys.exit(1)\n\n\ndef check_port_available(port: int) -> bool:\n \"\"\"Check that nothing is already listening on the local port.\n\n Returns True if the port is free, False if something is already bound.\n \"\"\"\n try:\n with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n s.bind((\"127.0.0.1\", port))\n return True\n except OSError:\n return False\n\n\ndef wait_for_port(port: int, timeout: float = 10.0) -> bool:\n \"\"\"Wait for the local port to become available (tunnel is up).\"\"\"\n deadline = time.monotonic() + timeout\n while time.monotonic() \u003c deadline:\n try:\n with socket.create_connection((\"127.0.0.1\", port), timeout=1.0):\n return True\n except (ConnectionRefusedError, OSError):\n time.sleep(0.5)\n return False\n\n\ndef start_tunnel(\n instance_id: str,\n cache_host: str,\n cache_port: int,\n local_port: int,\n region: str,\n profile: str | None,\n) -> None:\n \"\"\"Start SSM port forwarding and keep it running until interrupted.\"\"\"\n\n session = boto3.Session(profile_name=profile, region_name=region)\n\n # --- Pre-flight checks ---\n print(\"Pre-flight checks:\\n\")\n\n # 1. Port validation\n validate_port(cache_port, \"--cache-port\")\n validate_port(local_port, \"--local-port\")\n print(f\" [OK] Port numbers are valid (local={local_port}, remote={cache_port})\")\n\n # 2. Local port not already in use\n if not check_port_available(local_port):\n print(f\" [FAIL] Local port {local_port} is already in use.\")\n print(f\" Another process is listening on 127.0.0.1:{local_port}.\")\n print(f\" Stop that process first, or use --local-port to pick a different port.\")\n sys.exit(1)\n print(f\" [OK] Local port {local_port} is available\")\n\n # 3. AWS CLI\n if not check_aws_cli():\n print(\" [FAIL] AWS CLI not found on PATH.\")\n print(\" Install: https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html\")\n sys.exit(1)\n print(\" [OK] AWS CLI found\")\n\n # 4. SSM plugin\n if not check_ssm_plugin():\n print(\" [FAIL] Session Manager plugin not found on PATH.\")\n print(\" Install: https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-working-with-install-plugin.html\")\n sys.exit(1)\n print(\" [OK] Session Manager plugin found\")\n\n # 5. Instance running\n if not check_instance_running(instance_id, session):\n print(f\" [FAIL] Instance {instance_id} is not in 'running' state.\")\n print(\" Start the instance or choose a different one (use find_tunnel_host.py).\")\n sys.exit(1)\n print(f\" [OK] Instance {instance_id} is running\")\n\n # 6. SSM agent online\n if not check_instance_ssm_status(instance_id, session):\n print(f\" [WARN] Instance {instance_id} is not reporting to SSM as Online.\")\n print(\" The tunnel may fail. Ensure the instance has:\")\n print(\" - SSM agent installed and running\")\n print(\" - AmazonSSMManagedInstanceCore IAM policy attached\")\n print(\" - Network route to SSM endpoints (NAT gateway or VPC endpoint)\")\n print(\" Proceeding anyway ...\\n\")\n else:\n print(f\" [OK] Instance {instance_id} SSM agent is online\")\n\n print()\n\n # --- Build the SSM command ---\n parameters = json.dumps(\n {\n \"host\": [cache_host],\n \"portNumber\": [str(cache_port)],\n \"localPortNumber\": [str(local_port)],\n }\n )\n\n cmd = [\n \"aws\", \"ssm\", \"start-session\",\n \"--target\", instance_id,\n \"--document-name\", \"AWS-StartPortForwardingSessionToRemoteHost\",\n \"--parameters\", parameters,\n \"--region\", region,\n ]\n if profile:\n cmd.extend([\"--profile\", profile])\n\n print(f\"Starting tunnel: 127.0.0.1:{local_port} -> {cache_host}:{cache_port}\")\n print(f\" via SSM instance: {instance_id}\")\n print(f\" region: {region}\")\n print()\n print(\"Tunnel command:\")\n print(f\" {' '.join(cmd)}\")\n print()\n\n # --- Launch the tunnel ---\n proc = None\n try:\n proc = subprocess.Popen(\n cmd,\n stdout=subprocess.PIPE,\n stderr=subprocess.STDOUT,\n text=True,\n )\n\n # Give the tunnel a moment to start, then check if the port is up\n time.sleep(2)\n\n if proc.poll() is not None:\n # Process already exited -- something went wrong\n output = proc.stdout.read() if proc.stdout else \"\"\n print(f\"Tunnel process exited immediately (code {proc.returncode}):\\n\")\n print(output)\n sys.exit(1)\n\n if wait_for_port(local_port, timeout=15.0):\n print(f\"Tunnel is UP. Local port 127.0.0.1:{local_port} is accepting connections.\\n\")\n is_serverless = \".serverless.\" in cache_host\n if is_serverless:\n print(\"Note: ElastiCache Serverless requires TLS. Your client must connect\")\n print(f\" with TLS enabled and SNI set to {cache_host}\")\n print(f\" (the tunnel forwards raw TCP; TLS negotiation happens end-to-end).\\n\")\n print(\"Test with:\")\n print(f\" python scripts/test_connection.py 127.0.0.1 --port {local_port} \\\\\")\n print(f\" --tunnel-mode --server-name {cache_host} \\\\\")\n print(f\" --username \u003cuser> --password \u003cpassword>\")\n print()\n print(\"Press Ctrl+C to stop the tunnel.\")\n else:\n print(f\"Warning: local port {local_port} not yet responding after 15s.\")\n print(\"The tunnel may still be starting. Check the output below.\\n\")\n\n # Stream SSM output with periodic health checks.\n # Uses select() to avoid blocking forever if the session stalls.\n health_check_interval = 30 # seconds between port probes\n\n assert proc.stdout is not None\n while proc.poll() is None:\n ready, _, _ = select.select([proc.stdout], [], [], health_check_interval)\n if ready:\n line = proc.stdout.readline()\n if line:\n print(f\" [SSM] {line}\", end=\"\")\n else:\n break # EOF -- process closed stdout\n else:\n # No output for health_check_interval seconds, check tunnel\n if not wait_for_port(local_port, timeout=3.0):\n print(f\"\\n [WARN] Tunnel port {local_port} is no longer responding.\")\n print(\" The SSM session may have stalled. Press Ctrl+C to stop.\")\n\n proc.wait()\n if proc.returncode != 0:\n print(f\"\\nTunnel exited with code {proc.returncode}\")\n sys.exit(1)\n\n except KeyboardInterrupt:\n print(\"\\n\\nShutting down tunnel ...\")\n if proc and proc.poll() is None:\n proc.send_signal(signal.SIGTERM)\n try:\n proc.wait(timeout=5)\n except subprocess.TimeoutExpired:\n proc.kill()\n print(\"Tunnel closed.\")\n except Exception as exc:\n print(f\"\\nError: {exc}\")\n if proc and proc.poll() is None:\n proc.kill()\n sys.exit(1)\n\n\ndef main() -> None:\n parser = argparse.ArgumentParser(\n description=\"Start an SSM port-forwarding tunnel to an ElastiCache endpoint.\",\n formatter_class=argparse.RawDescriptionHelpFormatter,\n epilog=__doc__,\n )\n parser.add_argument(\n \"--instance-id\", required=True,\n help=\"SSM-managed EC2 instance ID (use find_tunnel_host.py to discover)\",\n )\n parser.add_argument(\n \"--cache-host\", required=True,\n help=\"ElastiCache endpoint hostname (e.g. my-cache.serverless.use1.cache.amazonaws.com)\",\n )\n parser.add_argument(\n \"--cache-port\", type=int, default=6379,\n help=\"ElastiCache port (default: 6379; serverless reader endpoint uses 6380)\",\n )\n parser.add_argument(\n \"--local-port\", type=int, default=6379,\n help=\"Local port to forward to (default: 6379)\",\n )\n _default_region = (\n os.environ.get(\"AWS_REGION\")\n or os.environ.get(\"AWS_DEFAULT_REGION\")\n or \"us-east-1\"\n )\n parser.add_argument(\n \"--region\", default=_default_region,\n help=f\"AWS region (default: {_default_region})\",\n )\n parser.add_argument(\n \"--profile\", default=None,\n help=\"AWS profile name\",\n )\n\n args = parser.parse_args()\n\n start_tunnel(\n instance_id=args.instance_id,\n cache_host=args.cache_host,\n cache_port=args.cache_port,\n local_port=args.local_port,\n region=args.region,\n profile=args.profile,\n )\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":11954,"content_sha256":"feb38dd48e537f6f891143547a9fffba4219662012ab5bcd27d20e66fe3cbafb"},{"filename":"scripts/test_connection.py","content":"#!/usr/bin/env python3\n\"\"\"\nElastiCache connection validator.\n\nUses the valkey-py client library to perform a proper TLS-enabled\nPING against an ElastiCache endpoint. Supports RBAC (username + password),\nIAM auth, and legacy AUTH tokens (node-based only).\n\nServerless caches REQUIRE TLS -- this script defaults to TLS-on.\n\nDependencies:\n pip install valkey\n pip install boto3 # only needed for --iam-auth (includes botocore)\n\nUsage examples:\n # Serverless with RBAC (TLS is default)\n python test_connection.py your-endpoint.serverless.use1.cache.amazonaws.com \\\\\n --username my-user --password my-password\n\n # Node-based with TLS and RBAC\n python test_connection.py your-endpoint.cache.amazonaws.com \\\\\n --username my-user --password my-password\n\n # Node-based with legacy AUTH token\n python test_connection.py your-endpoint.cache.amazonaws.com \\\\\n --password my-auth-token\n\n # Node-based WITHOUT TLS (not recommended)\n python test_connection.py your-endpoint.cache.amazonaws.com \\\\\n --no-tls --password my-auth-token\n\n # Serverless with IAM auth\n python test_connection.py your-endpoint.serverless.use1.cache.amazonaws.com \\\\\n --iam-auth --iam-user my-iam-user --region us-east-1 \\\\\n --cache-name my-cache\n\n # Through an SSM tunnel (see start_tunnel.py)\n python test_connection.py 127.0.0.1 --port 6379 \\\\\n --tunnel-mode --server-name my-cache.serverless.use1.cache.amazonaws.com \\\\\n --username my-user --password my-password\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport os\nimport ssl\nimport sys\nimport time\n\n# ---------------------------------------------------------------------------\n# Dependency check -- fail early with an actionable message instead of a\n# raw ImportError traceback.\n#\n# Required: valkey\n# Optional: boto3 -- only needed when using --iam-auth.\n# ---------------------------------------------------------------------------\ntry:\n import valkey as client_lib\n\n client_name = \"valkey-py\"\nexcept ImportError:\n print(\n \"Error: the 'valkey' package is required for this script.\\n\"\n \"Install it with:\\n\"\n \" pip install valkey\\n\"\n \"\\n\"\n \"If using IAM authentication, you also need:\\n\"\n \" pip install boto3\",\n file=sys.stderr,\n )\n sys.exit(2)\n\n\ndef get_iam_auth_token(\n cache_name: str, iam_user: str, region: str, is_serverless: bool = False\n) -> str:\n \"\"\"Generate a short-lived IAM auth token for ElastiCache.\"\"\"\n cache_name = cache_name.lower()\n try:\n import botocore.session\n from botocore.signers import RequestSigner\n from botocore.model import ServiceId\n except ImportError:\n print(\n \"Error: the 'botocore' package is required for IAM authentication.\\n\"\n \"Install it with:\\n\"\n \" pip install botocore\\n\"\n \"\\n\"\n \"You also need AWS credentials configured (via environment\\n\"\n \"variables, ~/.aws/credentials, or an IAM role).\",\n file=sys.stderr,\n )\n sys.exit(2)\n\n session = botocore.session.get_session()\n signer = RequestSigner(\n ServiceId(\"elasticache\"),\n region,\n \"elasticache\",\n \"v4\",\n session.get_credentials(),\n session.get_component(\"event_emitter\"),\n )\n url = signer.generate_presigned_url(\n {\n \"method\": \"GET\",\n \"url\": f\"http://{cache_name}/?Action=connect&User={iam_user}&ResourceType=ServerlessCache\" if is_serverless else f\"http://{cache_name}/?Action=connect&User={iam_user}\",\n \"body\": {},\n \"headers\": {},\n \"context\": {},\n },\n operation_name=\"connect\",\n expires_in=900,\n region_name=region,\n )\n token = url[len(\"http://\"):] if url.startswith(\"http://\") else url\n return token\n\n\ndef _check_dns(host: str) -> None:\n \"\"\"Warn if the host resolves to a private IP and is not localhost.\"\"\"\n import ipaddress\n import socket\n\n if host in (\"127.0.0.1\", \"localhost\", \"::1\"):\n return\n try:\n addr = socket.gethostbyname(host)\n ip = ipaddress.ip_address(addr)\n if ip.is_private:\n print(f\" WARNING: {host} resolves to private IP {addr}.\", file=sys.stderr)\n print(f\" If you are not in the VPC, this connection will time out.\", file=sys.stderr)\n print(f\" Use --tunnel-mode with scripts/start_tunnel.py for local development.\\n\", file=sys.stderr)\n except socket.gaierror:\n print(f\" WARNING: DNS lookup failed for {host}. The endpoint may be wrong,\", file=sys.stderr)\n print(f\" or you may not have DNS access to the VPC.\\n\", file=sys.stderr)\n\n\ndef test_connection(\n host: str,\n port: int,\n username: str | None,\n password: str | None,\n use_tls: bool,\n timeout: float,\n tunnel_mode: bool = False,\n server_name: str | None = None,\n is_serverless: bool = False,\n) -> bool:\n \"\"\"Connect to ElastiCache and run PING using the client library.\"\"\"\n _check_dns(host)\n tls_label = \"TLS\" if use_tls else \"plaintext\"\n if tunnel_mode:\n tls_label += \" tunnel-mode\"\n auth_label = (\n f\"user={username}\"\n if username\n else (\"password-auth\" if password else \"no-auth\")\n )\n print(f\"Connecting to {host}:{port} ({tls_label}, {auth_label}) via {client_name} ...\")\n\n kwargs = {\n \"host\": host,\n \"port\": port,\n \"socket_timeout\": timeout,\n \"socket_connect_timeout\": timeout,\n \"decode_responses\": True,\n }\n\n if use_tls:\n kwargs[\"ssl\"] = True\n if tunnel_mode:\n # Through an SSM tunnel the TLS cert is issued for the real cache\n # hostname (*.cache.amazonaws.com), not 127.0.0.1. Disable hostname\n # checking but keep certificate verification against the CA bundle.\n # Set SNI to the real cache hostname (--server-name) so that\n # serverless endpoints can route the TLS handshake correctly.\n kwargs[\"ssl_cert_reqs\"] = \"required\"\n kwargs[\"ssl_ca_certs\"] = ssl.get_default_verify_paths().cafile\n kwargs[\"ssl_check_hostname\"] = False\n if server_name:\n kwargs[\"ssl_server_hostname\"] = server_name\n print(f\" Tunnel mode: hostname verification disabled, target={host}:{port}\")\n print(f\" WARNING: tunnel mode is for LOCAL DEVELOPMENT ONLY\", file=sys.stderr)\n else:\n kwargs[\"ssl_cert_reqs\"] = \"required\"\n kwargs[\"ssl_ca_certs\"] = ssl.get_default_verify_paths().cafile\n\n if username:\n kwargs[\"username\"] = username\n if password:\n kwargs[\"password\"] = password\n\n # Detect serverless from hostname pattern OR the --serverless flag.\n # The flag is needed when connecting through a tunnel (e.g., 127.0.0.1)\n # where the hostname pattern is not present.\n is_serverless = is_serverless or \".serverless.\" in host\n\n # In tunnel mode, always use a non-cluster client. Cluster clients run\n # CLUSTER SLOTS/NODES discovery and try to connect directly to each node\n # using private VPC IPs, which are unreachable through the tunnel. A\n # standalone client connects only to the tunneled endpoint, which is\n # sufficient for a connectivity test (PING + INFO).\n use_cluster_client = is_serverless and not tunnel_mode\n\n try:\n if use_cluster_client:\n if client_name == \"valkey-py\":\n conn = client_lib.ValkeyCluster(**kwargs)\n else:\n conn = client_lib.RedisCluster(**kwargs)\n elif client_name == \"valkey-py\":\n conn = client_lib.Valkey(**kwargs)\n else:\n conn = client_lib.Redis(**kwargs)\n\n if is_serverless and tunnel_mode:\n print(\" Note: using standalone client through tunnel (cluster discovery\")\n print(\" would return unreachable VPC IPs). PING validates connectivity.\")\n except Exception as exc:\n print(f\" Client init FAILED: {exc}\", file=sys.stderr)\n return False\n\n # Step 1: PING\n start = time.monotonic()\n try:\n result = conn.ping()\n elapsed_ms = (time.monotonic() - start) * 1000\n print(f\" PING: OK ({elapsed_ms:.1f} ms)\")\n except client_lib.exceptions.AuthenticationError as exc:\n print(f\" PING FAILED: Authentication error -- {exc}\", file=sys.stderr)\n print(\" Hint: Check username/password or ensure the RBAC user exists and is active.\", file=sys.stderr)\n return False\n except client_lib.exceptions.ConnectionError as exc:\n msg = str(exc)\n print(f\" PING FAILED: Connection error -- {exc}\", file=sys.stderr)\n if \"SSL\" in msg or \"tls\" in msg.lower():\n if use_tls:\n print(\" Hint: TLS handshake failed. Verify the endpoint supports in-transit encryption.\", file=sys.stderr)\n else:\n print(\" Hint: This endpoint may require TLS. Retry without --no-tls (TLS is default).\", file=sys.stderr)\n else:\n print(\" Hint: Check security group rules, VPC placement, and whether you need a tunnel or jump host.\", file=sys.stderr)\n return False\n except Exception as exc:\n print(f\" PING FAILED: {exc}\", file=sys.stderr)\n return False\n\n # Step 2: INFO SERVER (lightweight metadata check)\n try:\n info = conn.info(\"server\")\n engine = info.get(\"redis_version\", info.get(\"valkey_version\", \"unknown\"))\n mode = info.get(\"redis_mode\", \"unknown\")\n print(f\" Server: engine version {engine}, mode={mode}\")\n except Exception:\n # INFO may be blocked by ACL; that is fine\n print(\" Server info: not available (ACL may restrict INFO command)\", file=sys.stderr)\n\n # Step 3: Verify TLS status\n if use_tls:\n print(\" TLS: enabled (in-transit encryption active)\")\n else:\n print(\" TLS: disabled -- consider enabling in-transit encryption for production use\", file=sys.stderr)\n\n try:\n conn.close()\n except Exception:\n pass\n\n print(\" Result: CONNECTION SUCCESSFUL\")\n return True\n\n\ndef main() -> None:\n parser = argparse.ArgumentParser(\n description=\"Validate connectivity to an ElastiCache endpoint (serverless or node-based).\",\n formatter_class=argparse.RawDescriptionHelpFormatter,\n epilog=__doc__,\n )\n parser.add_argument(\"host\", help=\"ElastiCache endpoint hostname\")\n parser.add_argument(\"--port\", type=int, default=6379, help=\"Port (default: 6379; serverless reader endpoint uses 6380)\")\n parser.add_argument(\"--timeout\", type=float, default=5.0, help=\"Connection timeout in seconds (default: 5)\")\n\n # TLS (default on)\n parser.add_argument(\n \"--no-tls\",\n action=\"store_true\",\n default=False,\n help=\"Disable TLS (not recommended; serverless caches REQUIRE TLS)\",\n )\n\n # Tunnel mode\n tunnel_group = parser.add_argument_group(\"tunnel mode (for SSM port-forwarded connections)\")\n tunnel_group.add_argument(\n \"--tunnel-mode\",\n action=\"store_true\",\n default=False,\n help=\"Enable tunnel mode: disables TLS hostname verification (dev only)\",\n )\n tunnel_group.add_argument(\n \"--server-name\",\n help=\"Real cache hostname for TLS SNI (required for serverless endpoints in tunnel mode)\",\n )\n\n # Auth\n auth_group = parser.add_argument_group(\"authentication\")\n auth_group.add_argument(\"--username\", help=\"RBAC username\")\n auth_group.add_argument(\"--password\", help=\"RBAC password or legacy AUTH token (node-based only)\")\n auth_group.add_argument(\n \"--iam-auth\",\n action=\"store_true\",\n default=False,\n help=\"Use IAM authentication (Valkey 7.2+ / Redis OSS 7.0+). Requires boto3.\",\n )\n auth_group.add_argument(\"--iam-user\", help=\"IAM-enabled ElastiCache user ID (required with --iam-auth)\")\n auth_group.add_argument(\"--cache-name\", help=\"Cache or replication group name (required with --iam-auth)\")\n auth_group.add_argument(\n \"--serverless\",\n action=\"store_true\",\n default=False,\n help=\"Target is a serverless cache (adds ResourceType=ServerlessCache to IAM token)\",\n )\n _default_region = os.environ.get(\"AWS_REGION\") or os.environ.get(\"AWS_DEFAULT_REGION\") or \"us-east-1\"\n auth_group.add_argument(\"--region\", default=_default_region, help=f\"AWS region (default: {_default_region})\")\n\n args = parser.parse_args()\n\n use_tls = not args.no_tls\n password = args.password\n username = args.username\n\n # Tunnel mode validation\n if args.tunnel_mode and args.no_tls:\n parser.error(\"--tunnel-mode requires TLS (cannot combine with --no-tls)\")\n\n # IAM auth flow\n if args.iam_auth:\n if not args.iam_user or not args.cache_name:\n parser.error(\"--iam-auth requires --iam-user and --cache-name\")\n print(f\"Generating IAM auth token for user={args.iam_user}, cache={args.cache_name} ...\")\n password = get_iam_auth_token(args.cache_name, args.iam_user, args.region, is_serverless=args.serverless)\n username = args.iam_user\n print(\" IAM token generated (valid for 15 minutes)\")\n\n success = test_connection(\n host=args.host,\n port=args.port,\n username=username,\n password=password,\n use_tls=use_tls,\n timeout=args.timeout,\n tunnel_mode=args.tunnel_mode,\n server_name=args.server_name,\n is_serverless=args.serverless,\n )\n sys.exit(0 if success else 1)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":13606,"content_sha256":"9937a3176ed6b025add4109fe5e5b6b52a82311c7ccbd1028edbf15edf5adb3c"},{"filename":"scripts/validate_references.py","content":"#!/usr/bin/env python3\n\"\"\"\nElastiCache Reference Validator\n\nValidates that all reference files have required metadata and follow\nproject conventions. Also checks scripts for module docstrings and\nCLI entrypoints.\n\nUsage:\n python scripts/validate_references.py # normal output\n python scripts/validate_references.py --verbose # show all PASS results too\n python scripts/validate_references.py --quiet # summary counts only\n\nExit codes:\n 0 -- all checks passed (warnings are OK)\n 1 -- one or more FAIL results\n\"\"\"\n\nimport argparse\nimport os\nimport re\nimport sys\n\n# ---------------------------------------------------------------------------\n# Constants\n# ---------------------------------------------------------------------------\n\nBASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))\nREFERENCES_DIR = os.path.join(BASE_DIR, \"references\")\nSCRIPTS_DIR = os.path.join(BASE_DIR, \"scripts\")\nSKILL_MD = os.path.join(BASE_DIR, \"SKILL.md\")\n\n# Minimum number of non-whitespace characters (beyond the title line) for a\n# reference file to be considered non-empty.\nMIN_CONTENT_CHARS = 50\n\n# Pattern for kebab-case filenames: lowercase letters, digits, hyphens.\nKEBAB_RE = re.compile(r\"^[a-z0-9]+(?:-[a-z0-9]+)*\\.md$\")\n\n\n# ---------------------------------------------------------------------------\n# Result tracking\n# ---------------------------------------------------------------------------\n\nclass Result:\n \"\"\"A single validation result.\"\"\"\n\n def __init__(self, level, filepath, message):\n self.level = level # \"PASS\", \"WARN\", or \"FAIL\"\n self.filepath = filepath # relative to BASE_DIR\n self.message = message\n\n def __str__(self):\n return f\"[{self.level}] {self.filepath}: {self.message}\"\n\n\n# ---------------------------------------------------------------------------\n# Helpers\n# ---------------------------------------------------------------------------\n\ndef _relative(path):\n \"\"\"Return a path relative to BASE_DIR for display.\"\"\"\n return os.path.relpath(path, BASE_DIR)\n\n\ndef _read_skill_md_references():\n \"\"\"Extract all reference file paths mentioned in SKILL.md.\"\"\"\n if not os.path.isfile(SKILL_MD):\n return set()\n with open(SKILL_MD, \"r\", encoding=\"utf-8\") as fh:\n content = fh.read()\n # Match backtick-quoted paths like `references/setup/engine-selection.md`\n return set(re.findall(r\"`(references/[a-zA-Z0-9/_-]+\\.md)`\", content))\n\n\n# ---------------------------------------------------------------------------\n# Reference file checks\n# ---------------------------------------------------------------------------\n\ndef validate_reference_file(filepath, skill_refs):\n \"\"\"Validate a single .md reference file. Returns a list of Result objects.\"\"\"\n results = []\n rel = _relative(filepath)\n filename = os.path.basename(filepath)\n\n # 1. Kebab-case filename\n if not KEBAB_RE.match(filename):\n results.append(Result(\"FAIL\", rel,\n f\"Filename '{filename}' does not follow kebab-case convention \"\n \"(expected lowercase letters, digits, and hyphens only)\"))\n else:\n results.append(Result(\"PASS\", rel, \"Filename follows kebab-case convention\"))\n\n # 2. Read file content\n try:\n with open(filepath, \"r\", encoding=\"utf-8\") as fh:\n content = fh.read()\n except Exception as exc:\n results.append(Result(\"FAIL\", rel, f\"Could not read file: {exc}\"))\n return results\n\n lines = content.split(\"\\n\")\n non_blank_lines = [ln for ln in lines if ln.strip()]\n\n # 3. Has a title (first non-blank line starts with #)\n if not non_blank_lines:\n results.append(Result(\"FAIL\", rel, \"File is completely empty\"))\n return results # no point checking further\n\n first_line = non_blank_lines[0].strip()\n if first_line.startswith(\"#\"):\n results.append(Result(\"PASS\", rel, f\"Has title: {first_line[:60]}\"))\n else:\n results.append(Result(\"FAIL\", rel,\n \"Missing title -- first non-blank line should start with '#'\"))\n\n # 4. Has meaningful content beyond the title\n # Strip the title line and count remaining non-whitespace characters\n body = \"\\n\".join(lines[1:]) if len(lines) > 1 else \"\"\n body_chars = len(re.sub(r\"\\s\", \"\", body))\n if body_chars \u003c MIN_CONTENT_CHARS:\n results.append(Result(\"FAIL\", rel,\n f\"File body has only {body_chars} non-whitespace chars \"\n f\"(minimum {MIN_CONTENT_CHARS}) -- appears empty or stub\"))\n else:\n results.append(Result(\"PASS\", rel,\n f\"File has meaningful content ({body_chars} body chars)\"))\n\n # 5. Referenced in SKILL.md (non-critical)\n # Build the expected reference path from the relative path (e.g. references/setup/foo.md)\n if rel in skill_refs:\n results.append(Result(\"PASS\", rel, \"Referenced in SKILL.md\"))\n else:\n results.append(Result(\"WARN\", rel, \"Not referenced in SKILL.md\"))\n\n return results\n\n\ndef validate_all_references():\n \"\"\"Walk references/ and validate every .md file.\"\"\"\n results = []\n skill_refs = _read_skill_md_references()\n\n if not os.path.isdir(REFERENCES_DIR):\n results.append(Result(\"FAIL\", \"references/\",\n \"references/ directory not found\"))\n return results, 0\n\n md_count = 0\n for root, _dirs, files in os.walk(REFERENCES_DIR):\n for fname in sorted(files):\n if not fname.endswith(\".md\"):\n continue\n md_count += 1\n filepath = os.path.join(root, fname)\n results.extend(validate_reference_file(filepath, skill_refs))\n\n if md_count == 0:\n results.append(Result(\"FAIL\", \"references/\",\n \"No .md files found under references/\"))\n\n return results, md_count\n\n\n# ---------------------------------------------------------------------------\n# Script file checks\n# ---------------------------------------------------------------------------\n\ndef validate_script_file(filepath):\n \"\"\"Validate a single .py script file. Returns a list of Result objects.\"\"\"\n results = []\n rel = _relative(filepath)\n\n try:\n with open(filepath, \"r\", encoding=\"utf-8\") as fh:\n content = fh.read()\n except Exception as exc:\n results.append(Result(\"FAIL\", rel, f\"Could not read file: {exc}\"))\n return results\n\n # 1. Has a module docstring\n # A module docstring is a triple-quoted string near the top of the file,\n # possibly preceded by a shebang, encoding declaration, or comments.\n # We use a simple heuristic: check if '\"\"\"' or \"'''\" appears in the first\n # 20 non-blank lines.\n lines = content.split(\"\\n\")\n top_chunk = \"\\n\".join(lines[:30])\n has_docstring = ('\"\"\"' in top_chunk or \"'''\" in top_chunk)\n\n if has_docstring:\n results.append(Result(\"PASS\", rel, \"Has module docstring\"))\n else:\n results.append(Result(\"FAIL\", rel,\n \"Missing module docstring at top of file\"))\n\n # 2. Has a CLI entrypoint (if __name__ == \"__main__\")\n has_main = ('if __name__' in content\n and ('\"__main__\"' in content or \"'__main__'\" in content))\n if has_main:\n results.append(Result(\"PASS\", rel, \"Has CLI entrypoint\"))\n else:\n # This is a warning, not a fail, because some scripts are libraries\n results.append(Result(\"WARN\", rel,\n \"No 'if __name__ == \\\"__main__\\\"' block -- \"\n \"OK if this is a library, not a CLI tool\"))\n\n return results\n\n\ndef validate_all_scripts():\n \"\"\"Walk scripts/ and validate every .py file (excluding this script).\"\"\"\n results = []\n\n if not os.path.isdir(SCRIPTS_DIR):\n results.append(Result(\"FAIL\", \"scripts/\",\n \"scripts/ directory not found\"))\n return results, 0\n\n py_count = 0\n for fname in sorted(os.listdir(SCRIPTS_DIR)):\n if not fname.endswith(\".py\"):\n continue\n py_count += 1\n filepath = os.path.join(SCRIPTS_DIR, fname)\n results.extend(validate_script_file(filepath))\n\n if py_count == 0:\n results.append(Result(\"WARN\", \"scripts/\",\n \"No .py files found under scripts/\"))\n\n return results, py_count\n\n\n# ---------------------------------------------------------------------------\n# Reporting\n# ---------------------------------------------------------------------------\n\ndef print_results(results, verbose=False, quiet=False):\n \"\"\"Print validation results to stdout.\"\"\"\n fail_count = sum(1 for r in results if r.level == \"FAIL\")\n warn_count = sum(1 for r in results if r.level == \"WARN\")\n pass_count = sum(1 for r in results if r.level == \"PASS\")\n\n if not quiet:\n # Group results by level for readability\n if fail_count > 0:\n print(\"\\n--- FAILURES ---\")\n for r in results:\n if r.level == \"FAIL\":\n print(f\" {r}\")\n\n if warn_count > 0:\n print(\"\\n--- WARNINGS ---\")\n for r in results:\n if r.level == \"WARN\":\n print(f\" {r}\")\n\n if verbose and pass_count > 0:\n print(\"\\n--- PASSED ---\")\n for r in results:\n if r.level == \"PASS\":\n print(f\" {r}\")\n\n # Summary line (always printed)\n print(f\"\\nSummary: {pass_count} passed, {warn_count} warnings, {fail_count} failures\")\n\n return fail_count\n\n\n# ---------------------------------------------------------------------------\n# Main\n# ---------------------------------------------------------------------------\n\ndef main():\n parser = argparse.ArgumentParser(\n description=\"Validate ElastiCache reference files and scripts.\"\n )\n parser.add_argument(\n \"--verbose\", action=\"store_true\",\n help=\"Show PASS results in addition to WARN and FAIL\"\n )\n parser.add_argument(\n \"--quiet\", action=\"store_true\",\n help=\"Show summary counts only (no individual results)\"\n )\n args = parser.parse_args()\n\n all_results = []\n\n # Validate reference files\n print(\"Validating reference files...\")\n ref_results, ref_count = validate_all_references()\n all_results.extend(ref_results)\n print(f\" Scanned {ref_count} reference .md files\")\n\n # Validate scripts\n print(\"Validating scripts...\")\n script_results, script_count = validate_all_scripts()\n all_results.extend(script_results)\n print(f\" Scanned {script_count} script .py files\")\n\n # Print results\n fail_count = print_results(all_results, verbose=args.verbose, quiet=args.quiet)\n\n sys.exit(1 if fail_count > 0 else 0)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":10850,"content_sha256":"ebcca22b0533e71c002e259202686a187c0a79f0d38efe012228598bfc64d1f3"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"ElastiCache","type":"text"}]},{"type":"paragraph","content":[{"text":"A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"How this skill works","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording (\"help me figure out which data structures to use\" matches ","type":"text"},{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]},{"text":" even without the word \"pattern\").","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Disambiguation:","type":"text","marks":[{"type":"strong"}]},{"text":" If the user's intent matches multiple sub-skills, apply these rules in order:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If ","type":"text"},{"text":".elasticache/requirements.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" exists with ","type":"text"},{"text":"infrastructure.endpoint","type":"text","marks":[{"type":"code_inline"}]},{"text":" set, prefer ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]},{"text":" (the user has an existing cache).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If no cache exists (no requirements.json or no endpoint), prefer ","type":"text"},{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If still ambiguous, ask one clarifying question: \"Are you looking to set up something new, or troubleshoot something existing?\"","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Check the Guardrails section before recommending an engine or deployment model.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"references/{sub-skill-id}/instructions.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the matched sub-skill. If the file is not found at a relative path, check your prompt or environment for the skill directory absolute path and retry with ","type":"text"},{"text":"{skill-directory}/references/{sub-skill-id}/instructions.md","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If the request spans multiple sub-skills, execute them in pipeline order.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If a sub-skill requires upstream context (engine, deployment model, endpoint) not yet in session memory, route to the upstream sub-skill first.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If no sub-skill matches, activate ","type":"text"},{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":" first.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If a script or CLI call fails, show the error to the user and suggest a specific fix before retrying.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Sub-skill registry","type":"text"}]},{"type":"paragraph","content":[{"text":"Each entry has: an ID (directory name under ","type":"text"},{"text":"references/","type":"text","marks":[{"type":"code_inline"}]},{"text":"), a domain description, semantic categories for matching, and upstream/downstream dependencies.","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"ID","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Name","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Domain","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Semantic Categories","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Upstream","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Downstream","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Solution Fit","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Gathers workload, stack, scale, latency, persistence, and budget through workspace scan + structured interview. Decides whether ElastiCache is the right service and hands off with a routing recommendation.","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"I need a cache, speed up my app, reduce database load, lower Bedrock cost, should I use ElastiCache, what's best for my workload, evaluating cache options, ElastiCache vs X, Valkey vs X, vague new workload","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"—","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"genai","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"migration","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Create and Connect","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Provisioning, connectivity, security, authentication, IaC, deployment choice. Gets the user to a working cache with least friction. Covers engine selection, serverless vs node-based, VPC, TLS, RBAC/IAM, jump-host/SSM tunnels, CLI/SDK/CFN/CDK/Terraform starters.","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"create a cache, set up ElastiCache, provision, Valkey cluster, connect Lambda/ECS/EKS/EC2, VPC, security groups, TLS, RBAC, IAM auth, jump host, SSM tunnel, CloudFormation, CDK, Terraform, engine selection, serverless vs node-based, backup, snapshot, restore, export","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":" (optional)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"genai","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Application Patterns","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Picks data structures, key schema, TTL strategy, invalidation approach, and client code for non-AI patterns: cache-aside, session store, rate limiting, leaderboards, counters, pub/sub, streams, shopping carts, job queues, activity feeds.","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"session store, rate limiting, leaderboard, cache-aside, query caching, counters, streams, pub/sub, shopping cart, job queue, activity feed, key schema, TTL, invalidation, data structures","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":" (cache must exist)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"genai","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"AI and Vector Workloads","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Classifies request into Mode 1 (plain cache), Mode 2 (semantic response cache), or Mode 3 (full vector search). Selects Valkey and forces node-based Valkey 8.2 or above (recommend 9.0) when server-side vector similarity is needed. Covers semantic caching, agent memory, RAG retrieval, recommendation, personalization, conversation/session persistence for AI agents, and framework wiring (Strands, mem0, LangChain).","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"semantic cache, RAG, agent memory, conversational memory, vector search, embeddings, recommendation, personalization, Bedrock latency, Bedrock cost, LLM caching, Strands, mem0, LangChain, conversation history, AI session store, embedding provider, framework integration","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":" (cache must exist)","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Operate and Observe","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Diagnoses performance, cost, and reliability using metrics first, then recommends the smallest change. Covers dashboards, alarms, log delivery, cost reporting, event routing, troubleshooting high CPU / memory / replication lag / connection spikes / low hit rate / hot keys / big keys / slot imbalance / latency spike root cause.","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"cache is slow, cost too high, hit rate low, high CPU, memory pressure, replication lag, connection spikes, dashboards, alarms, CloudWatch, cost comparison, troubleshoot, hot key, uneven shard load, one node pinned, big key, memory bloat, which key is biggest, keyspace distribution, prefix analysis, cost attribution by tenant, memory imbalance, one shard full, slot memory skew, latency spike, slow command incident, root cause for latency bump","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"—","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"migration","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"migration","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Engine and Platform Migration","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Selects the migration path and sequences preflight, validation, cutover, and rollback. Covers self-managed Redis → ElastiCache, Redis OSS → Valkey, node-based ↔ serverless, version upgrades. Hard validate-before-migrate gate.","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"migrate, Redis OSS to Valkey, self-managed to ElastiCache, node-based to serverless, serverless to node-based, engine upgrade, version upgrade, zero-downtime cutover, rollback","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"—","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Pipeline order","type":"text"}]},{"type":"paragraph","content":[{"text":"Sub-skills run independently, but common multi-step journeys follow these pipelines:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":" → (","type":"text"},{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]},{"text":" | ","type":"text"},{"text":"genai","type":"text","marks":[{"type":"code_inline"}]},{"text":") → ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"migration","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":" | ","type":"text"},{"text":"migration","type":"text","marks":[{"type":"code_inline"}]},{"text":" (if metrics indicate)","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"State handoff: requirements.json","type":"text"}]},{"type":"paragraph","content":[{"text":".elasticache/requirements.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" is the single source of truth for cross-sub-skill state. Each sub-skill reads it at start and writes its section after completing work. Read before writing; merge, do not overwrite.","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Section","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Owner","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Key fields","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"top-level","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"engine","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"deployment_model","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"region","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"runtime","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"patterns","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"use_case","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"vpc_id","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"subnet_ids","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"security_group_ids","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"infrastructure","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"cache_name","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"resource_id","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"engine_version","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"topology","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"endpoint","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"port","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"auth_model","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"tls","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"client_library","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"execution_path","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"access_mode","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"tunnel_instance_id","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"embedding_provider","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"embedding_model","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"embedding_dim","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"embedding_module","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"genai","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"genai","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mode","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"mode_2_path","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"framework","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"migration","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"migration","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"source_type","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"source_host","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"migration_path","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"cutover_status","type":"text","marks":[{"type":"code_inline"}]}]}]}]}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Ownership note:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"deployment_model","type":"text","marks":[{"type":"code_inline"}]},{"text":" is set by ","type":"text"},{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":" during initial interview. ","type":"text"},{"text":"migration","type":"text","marks":[{"type":"code_inline"}]},{"text":" may update it after an engine or deployment model switch (e.g., node-based to serverless).","type":"text"}]}]},{"type":"paragraph","content":[{"text":"requirements.json should include ","type":"text"},{"text":"\"schema_version\": 1","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"\"last_updated\": \"\u003cISO timestamp>\"","type":"text","marks":[{"type":"code_inline"}]},{"text":" at the top level. Every sub-skill that writes to requirements.json must update ","type":"text"},{"text":"last_updated","type":"text","marks":[{"type":"code_inline"}]},{"text":". If ","type":"text"},{"text":"last_updated","type":"text","marks":[{"type":"code_inline"}]},{"text":" is older than 7 days, warn the user that cached state may be stale.","type":"text"}]},{"type":"paragraph","content":[{"text":"requirements.json tracks one active cache. If the user works with multiple caches in the same project, confirm which cache is active before reading or writing state.","type":"text"}]},{"type":"paragraph","content":[{"text":"When a sub-skill needs upstream context (engine, endpoint, auth model), check requirements.json first. If the field is ","type":"text"},{"text":"null","type":"text","marks":[{"type":"code_inline"}]},{"text":" or the file does not exist, route to the upstream sub-skill.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Global rules (apply to every sub-skill)","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Execution path.","type":"text","marks":[{"type":"strong"}]},{"text":" Use AWS CLI, SDK (boto3), CloudFormation, or CDK as the primary path for control-plane work. Use valkey-py as the primary path for data-plane work.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Response depth.","type":"text","marks":[{"type":"strong"}]},{"text":" Summary (2-3 sentences) for \"should I\" or \"which\" questions. Standard (recommendation + config + code + next steps) by default. Expert (full decision matrix with alternatives, cost, security caveats) for \"why\" or \"compare all\" questions. Escalate on user request; never downgrade unprompted.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Session memory.","type":"text","marks":[{"type":"strong"}]},{"text":" Track region, VPC, engine, deployment model, auth model, compute runtime, and language. Carry forward across sub-skills. Do not re-ask. If the user overrides a value, update it everywhere. Inferred values (from workspace scan or IaC) must be re-confirmed before high-risk decisions (engine, deployment model, security posture); low-risk inferences (language, framework, region) can be used as defaults silently.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Source priority.","type":"text","marks":[{"type":"strong"}]},{"text":" Always answer from skill-local files first (sub-skill references, then ","type":"text"},{"text":"scripts/","type":"text","marks":[{"type":"code_inline"}]},{"text":"). Do not fetch external documentation or web search unless the local files cannot answer the query. When local files are insufficient, fall back to official AWS docs: https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/ for features and https://aws.amazon.com/elasticache/pricing/ for pricing. Never invent price points or version constraints. If the user references a Valkey or Redis version, feature, or pricing tier not covered in local files, fall back to https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/ before answering. Do not extrapolate from local content that may be outdated.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Freshness disclaimer.","type":"text","marks":[{"type":"strong"}]},{"text":" When outputting pricing, version constraints, or feature availability, include a one-line disclaimer: \"For current pricing see https://aws.amazon.com/elasticache/pricing/. For current feature availability see https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/.\"","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Connection safety.","type":"text","marks":[{"type":"strong"}]},{"text":" Never create a Valkey/Redis client at module level (top of file, import time). Initialize connections inside a function or on first use. Module-level connections crash applications that import the module before the cache is reachable.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Code environment.","type":"text","marks":[{"type":"strong"}]},{"text":" Use ","type":"text"},{"text":"python3","type":"text","marks":[{"type":"code_inline"}]},{"text":" (not ","type":"text"},{"text":"python","type":"text","marks":[{"type":"code_inline"}]},{"text":") in all commands and shebangs. Do not generate ","type":"text"},{"text":"pip install","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"pytest","type":"text","marks":[{"type":"code_inline"}]},{"text":" commands in application code. Prefer the agent's file-writing tool over shell redirection for writing files.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Input validation (write operations only).","type":"text","marks":[{"type":"strong"}]},{"text":" Before running any create, modify, or delete CLI command or SDK call that uses user-provided values (cache name, region, subnet IDs, security group IDs, VPC ID, KMS key, snapshot name), validate them with ","type":"text"},{"text":"scripts/input_validator.py","type":"text","marks":[{"type":"code_inline"}]},{"text":". Run ","type":"text"},{"text":"python3 scripts/input_validator.py cache_name=\u003cval> region=\u003cval> ...","type":"text","marks":[{"type":"code_inline"}]},{"text":" or import ","type":"text"},{"text":"validate_all","type":"text","marks":[{"type":"code_inline"}]},{"text":" as a library. Skip validation for read-only operations (describe, list) where bad input is harmless.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Resource tagging (always apply on resource creation).","type":"text","marks":[{"type":"strong"}]},{"text":" When creating any cache (","type":"text"},{"text":"create-serverless-cache","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"create-replication-group","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"create-cache-cluster","type":"text","marks":[{"type":"code_inline"}]},{"text":"), ALWAYS include these tags:","type":"text"}]},{"type":"paragraph","content":[{"text":"--tags Key=managed_by,Value=aws-skills Key=skill,Value=elasticache Key=skill_version,Value=1.0.0 Key=created_by,Value=elasticache-skill Key=generation_model,Value={your-model-id}","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"Example: ","type":"text"},{"text":"--tags Key=managed_by,Value=aws-skills Key=skill,Value=elasticache Key=skill_version,Value=1.0.0 Key=created_by,Value=elasticache-skill Key=generation_model,Value=claude-sonnet-4-20250514","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"Set ","type":"text"},{"text":"generation_model","type":"text","marks":[{"type":"code_inline"}]},{"text":" to your own model identifier. Include these tags even if the user does not mention tagging. If the user provides additional tags, append these to their tags rather than replacing them.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Reference loading","type":"text"}]},{"type":"paragraph","content":[{"text":"Load additional references only when the current turn's answer requires them.","type":"text"}]},{"type":"paragraph","content":[{"text":"On-demand pointers (not preloaded; fetch when the trigger applies):","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-ux/production-readiness.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when the user asks if their cache is ready for production, or after setup completes and the user wants to go to production","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-ux/action-safety.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — before any destructive action (risk levels, never-auto-execute list)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-ux/error-remediation.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when the user hits a specific ElastiCache error code (MOVED, CROSSSLOT, CLUSTERDOWN, MULTI/EXEC+IAM, etc.)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-foundation/boundary-doc.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when the user asks what this skill covers","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-foundation/attribution.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when generating CLI commands, SDK code, or IaC templates","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-foundation/architecture-diagrams.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when the user asks for architecture diagrams or visual reference","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-runtime/lambda.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when connecting from Lambda (cold start gotchas, IAM auth code, lazy init)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-runtime/ecs.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when connecting from ECS (SIGTERM shutdown, connection pool drain, task definition)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-runtime/eks.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when connecting from EKS (IRSA, service mesh bypass, SecurityGroupPolicy CRD)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-runtime/api-gateway.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when integrating with API Gateway (no direct path, caching layers comparison)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-runtime/rds-acceleration.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when caching RDS/Aurora queries (thundering herd, stampede protection, invalidation)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-runtime/secret-injection.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when the user asks about credential management per compute platform","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-security/encryption-defaults.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when adding encryption to an existing unencrypted cluster (TLS two-step migration, at-rest immutability)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-security/config-guardrails.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when the user wants continuous compliance monitoring (AWS Config rules, custom Lambda rules)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/shared-security/vpc-patterns.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — when debugging port/security-group issues (port 6380 serverless reader, anti-patterns)","type":"text"}]}]}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Folder convention:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"references/","type":"text","marks":[{"type":"code_inline"}]},{"text":" contains 10 folders. 6 match the sub-skills (","type":"text"},{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"data-modeling","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"genai","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"monitoring","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"migration","type":"text","marks":[{"type":"code_inline"}]},{"text":") and are routing destinations. The 4 ","type":"text"},{"text":"shared-*","type":"text","marks":[{"type":"code_inline"}]},{"text":" folders (","type":"text"},{"text":"shared-foundation","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"shared-ux","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"shared-security","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"shared-runtime","type":"text","marks":[{"type":"code_inline"}]},{"text":") are cross-cutting material loaded on demand, not routing destinations.","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Guardrails","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Priority","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Rule","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"CRITICAL","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Vector search MUST use node-based Valkey 8.2 or above.","type":"text","marks":[{"type":"strong"}]},{"text":" Serverless does NOT support vector search. Never suggest serverless for vector search. Apply this regardless of which sub-skill activates.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"CRITICAL","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" invent price points or version constraints. Use ","type":"text"},{"text":"scripts/price_calculator.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" and current AWS docs when precision matters.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"HIGH","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" recommend Memcached when the user needs persistence, replication, RBAC or IAM auth, sorted sets, streams, pub/sub, or vector search.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"HIGH","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" assume local laptop access works directly. ElastiCache is VPC-centric; explain VPC, tunnel, or jump-host access when needed.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"STANDARD","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" trigger on every generic Redis mention. Trigger when the user is clearly asking about AWS, managed caching, migration, connectivity, pricing, operations, or AWS service integration.","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"STANDARD","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"For ambiguous \"cache\" requests inside AWS contexts, activate this skill and start with ","type":"text"},{"text":"requirements","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Product truths","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ElastiCache Serverless deploys in under a minute and removes infrastructure management.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Valkey serverless pricing is 33% lower than other supported engines; node-based Valkey pricing is 20% lower.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Serverless caches have in-transit encryption always enabled (cannot be disabled).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"IAM auth is available for all ElastiCache Valkey versions (7.2 is the baseline Valkey version on ElastiCache) and Redis OSS 7.0+.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Valkey version ladder: 7.2 (baseline), 8.0 (20% more data per node (capacity improvement), per-slot metrics), 8.1 (Bloom filters, COMMANDLOG, SET IFEQ, 20% less memory via new hash table (efficiency improvement)), 8.2 (vector search), 9.0 (recommended default for new clusters). Recommend Valkey 9.0 for new clusters unless a specific feature dictates otherwise.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Vector search is available for Valkey 8.2 or above on node-based clusters (recommend 9.0).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Global Datastore is available for node-based clusters only. It does not support IPv6 or Local Zones. Global Datastore supports AUTH and RBAC. Cross-region failover must be promoted manually (no autofailover across regions). At-rest encryption must be enabled on all clusters in the Global Datastore, but each cluster can use a separate KMS key per region.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Online migration from self-managed Redis to ElastiCache requires: (source) AUTH must not be enabled, ","type":"text"},{"text":"protected-mode","type":"text","marks":[{"type":"code_inline"}]},{"text":" set to ","type":"text"},{"text":"no","type":"text","marks":[{"type":"code_inline"}]},{"text":", replication and administrative commands must not be renamed (e.g., ","type":"text"},{"text":"sync","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"psync","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"info","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"config","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"command","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"cluster","type":"text","marks":[{"type":"code_inline"}]},{"text":"); (target) encryption in-transit disabled, Multi-AZ enabled, engine version Redis OSS 5.0.6+ or Valkey 7.2+, not part of a Global Datastore, data tiering disabled. Shard counts must match between source and target. All source Redis instances must use the same port. Online migration is not supported for serverless caches (node-based targets only). See ","type":"text"},{"text":"references/migration/topology-validation.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the full checklist.","type":"text"}]}]}]}]},"metadata":{"date":"2026-11-02","name":"amazon-elasticache","author":"@skillopedia","source":{"stars":3376,"repo_name":"agent-toolkit-for-aws","origin_url":"https://github.com/aws/agent-toolkit-for-aws/blob/HEAD/skills/specialized-skills/database-skills/amazon-elasticache/SKILL.md","repo_owner":"aws","body_sha256":"1cfa793c277f309c989d9007d021d44206d968889042bf95edd1aaa82bc1c3f8","cluster_key":"dc0bd93269ce662c2d4aab0ee4dacab5249f2fc3baf8380675d7e5fedfc0ed61","clean_bundle":{"format":"clean-skill-bundle-v1","source":"aws/agent-toolkit-for-aws/skills/specialized-skills/database-skills/amazon-elasticache/SKILL.md","text_bytes":766575,"source_commit":"1761e96813ebcbd43ed81d19e837219033f0afe5","skills_sh_rank":4000,"attachment_count":79,"text_attachments":79,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/specialized-skills/database-skills/amazon-elasticache/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"data-analytics","category_label":"Data"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"data-analytics","import_tag":"skills-sh-top20k-2026-11","description":"Activate when developers have latent caching needs: slow API responses, database read bottlenecks, DynamoDB throttling or cost, RDS/Aurora scaling pressure, Bedrock latency or cost, or adding a cache; activate when working with Redis, Valkey, Memcached, or any in-memory data store, cache-aside patterns, session stores, rate limiting, leaderboards, counters, streams, queues, pub/sub, distributed locks, feature flags, shopping carts, or other caching strategies. Activate for GenAI and ML retrieval: vector similarity search for low-latency retrieval, semantic caching, RAG, LLM response caching, embedding stores, AI agent memory, recommendation, personalization. Activate for ElastiCache lifecycle: provisioning (serverless or node-based), engine selection, CloudFormation/CDK/Terraform IaC, VPC connectivity, TLS, RBAC, IAM auth, Global Datastore, monitoring, troubleshooting, cost optimization, and migration from self-managed Redis. Do not trigger for browser caches, CDN/CloudFront, HTTP Cache-Control, CPU caches."}},"renderedAt":1793666209898}

ElastiCache A modular ElastiCache toolkit organized as a registry of sub-skills. Each sub-skill handles one domain of ElastiCache work. The router below matches user intent to the right sub-skill, then loads only the references needed for that sub-skill. How this skill works 1. Match the user's request against the semantic categories in the registry below. Match on meaning, not exact wording ("help me figure out which data structures to use" matches even without the word "pattern"). 2. Disambiguation: If the user's intent matches multiple sub-skills, apply these rules in order: - If exists wi…