Coverage for src/tagmania/snapshot_manager.py: 98%
125 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-02 01:51 +0000
1"""AWS Cluster Snapshot Management CLI.
3This module provides the main CLI interface for creating, restoring, deleting,
4and listing EBS snapshots for EC2 clusters. It supports both full cluster
5operations and targeted operations using regex patterns to match specific instances.
7The tool relies on EC2 instances having "Cluster" tags to identify cluster membership
8and "Name" tags for targeted operations. All operations include safety features
9with confirmation prompts to prevent accidental data loss.
11Features:
12 - Create named snapshots of entire clusters
13 - Restore clusters from snapshots with volume replacement
14 - Targeted restore using regex patterns on instance names
15 - List and delete existing snapshots
16 - Safety confirmations for all destructive operations
18Usage:
19 The module is typically invoked via the cluster-snap CLI command:
21 ```bash
22 # Create a backup
23 cluster-snap --backup --name daily-backup production
25 # Restore entire cluster
26 cluster-snap --restore --name daily-backup production
28 # Targeted restore (only web servers)
29 cluster-snap --restore --target ".*-web-.*" production
31 # List snapshots
32 cluster-snap --list production
34 # Delete snapshots
35 cluster-snap --delete --name daily-backup production
36 ```
38Warning:
39 Restore operations permanently delete existing EBS volumes and replace them
40 with volumes created from snapshots. This operation cannot be undone.
41 Always confirm you have the correct backup before proceeding.
42"""
44import argparse
45import logging
46import re
48from tagmania.iac_tools.clusterset import ClusterSet
49from tagmania.iac_tools.timing import log_duration
52def _configure_logging() -> logging.Logger:
53 """Attach a stderr handler to the tagmania logger so INFO lines show up in the CLI."""
54 logger = logging.getLogger("tagmania")
55 if not logger.handlers:
56 handler = logging.StreamHandler()
57 handler.setFormatter(logging.Formatter("%(message)s"))
58 logger.addHandler(handler)
59 logger.setLevel(logging.INFO)
60 return logger
63def main():
64 """Main entry point for the cluster snapshot management CLI.
66 Parses command line arguments and executes the appropriate snapshot operation
67 (backup, restore, list, or delete) on the specified cluster.
69 The function handles all user interactions including confirmation prompts
70 for destructive operations and provides detailed feedback on operation progress.
72 Raises:
73 SystemExit: On invalid command line arguments or user cancellation.
74 ValueError: On invalid regex patterns for targeted operations.
75 AWSError: On AWS API failures during snapshot operations.
76 """
77 parser = argparse.ArgumentParser(
78 description="AWS cluster snapshot backup and restore tool.",
79 epilog="""
80 This tool relies on the "Cluster" and "Owner" tags on instances and
81 volumes. IAC automation puts this in place. This tool
82 creates multiple sets of labeled snapshots and creates volumes from
83 them when a restore operation is performed.""",
84 )
85 group = parser.add_mutually_exclusive_group(required=True)
86 group.add_argument(
87 "-b",
88 "--backup",
89 action="store_const",
90 dest="backup",
91 const=True,
92 default=False,
93 help="Create snapshots for cluster CLUSTER.",
94 )
95 group.add_argument(
96 "-D",
97 "--delete",
98 action="store_const",
99 dest="delete",
100 const=True,
101 default=False,
102 help="Delete snapshots from cluster CLUSTER.",
103 )
104 group.add_argument(
105 "-r",
106 "--restore",
107 action="store_const",
108 dest="restore",
109 const=True,
110 default=False,
111 help="Restore cluster CLUSTER from snapshots.",
112 )
113 group.add_argument(
114 "-l",
115 "--list",
116 action="store_const",
117 dest="list",
118 const=True,
119 default=False,
120 help="List snapshots with a given label or all if none specified.",
121 )
122 parser.add_argument("-n", "--name", default=None, help="Name to use for the snapshots.")
123 parser.add_argument(
124 "-t",
125 "--target",
126 type=str,
127 default=None,
128 help="Regex pattern to match instance Name tags for targeted restore.",
129 )
130 parser.add_argument(
131 "cluster",
132 help="""
133 the name CLUSTER of the cluster in question. This can be found by
134 looking at any node in the AWS console and looking for the "Cluster"
135 tag.""",
136 )
137 parser.add_argument("--profile", "-p", help="the AWS profile to use", default=None)
138 args = parser.parse_args()
140 logger = _configure_logging()
141 cluster = ClusterSet(args.cluster, profile=args.profile)
143 if args.backup:
144 snapshot_name = "default" if args.name is None else args.name
145 confirm = input(f"Create backup of {args.cluster} named '{snapshot_name}'? [no] ")
146 if confirm == "yes":
147 print("Making backup.")
148 instances = cluster.get_instances()
149 if len(instances) == 0:
150 print("No instances found. Operation aborted.")
151 else:
152 with log_duration(logger, "backup"):
153 # Stop the cluster (not clean)
154 cluster.stop_instances()
155 cluster.create_snapshots(snapshot_name)
156 # Start cluster
157 # cluster.start_instances()
158 print("Operation completed successfully!")
159 else:
160 print("Operation aborted.")
162 if args.delete:
163 if args.name is None:
164 snapshot_name = "*"
165 confirm_string = f"Delete all backups for {args.cluster}"
166 else:
167 snapshot_name = args.name
168 confirm_string = f"Delete backup of {args.cluster} named '{snapshot_name}'"
169 confirm = input(f"{confirm_string}? [no] ")
170 if confirm == "yes":
171 print("Deleting snapshots.")
172 snapshots = cluster.get_snapshots(snapshot_name)
173 if len(snapshots) == 0:
174 print("No snapshots found. Operation aborted.")
175 else:
176 cluster.delete_snapshots(snapshot_name)
177 else:
178 print("Operation aborted.")
180 if args.restore:
181 snapshot_name = "default" if args.name is None else args.name
183 # Handle targeted restore
184 if args.target:
185 try:
186 # Validate regex pattern
187 re.compile(args.target)
189 # Check if any instances match the pattern
190 instances = cluster.get_instances()
191 filtered_instances = cluster._filter_instances_by_name_regex(instances, args.target)
193 if len(filtered_instances) == 0:
194 print(
195 f"No instances found matching pattern '{args.target}'. Operation aborted."
196 )
197 else:
198 print(
199 f"Found {len(filtered_instances)} instances matching pattern '{args.target}':"
200 )
201 for instance in filtered_instances:
202 name_tag = "Unknown"
203 if instance.tags:
204 for tag in instance.tags:
205 if tag["Key"] == "Name":
206 name_tag = tag["Value"]
207 break
208 print(f" - {instance.id} ({name_tag})")
210 confirm = input(
211 f"Restore backup '{snapshot_name}' for these {len(filtered_instances)} instances? [no] "
212 )
213 if confirm == "yes":
214 print("Restoring targeted instances.")
215 with log_duration(logger, "restore_targeted"):
216 # Stop targeted instances
217 cluster.stop_instances_targeted(args.target)
218 # Detach and delete volumes from targeted instances
219 cluster.detach_volumes_targeted(args.target)
220 cluster.delete_volumes_targeted(args.target)
221 # Create new volumes from snapshots and attach them
222 cluster.create_volumes_targeted(snapshot_name, args.target)
223 cluster.attach_volumes_targeted(snapshot_name, args.target)
224 # Start targeted instances
225 # cluster.start_instances_targeted(args.target)
226 print("Operation completed successfully!")
227 else:
228 print("Operation aborted.")
229 except re.error as e:
230 print(f"Invalid regex pattern '{args.target}': {e}")
231 print("Operation aborted.")
232 else:
233 # Full cluster restore
234 confirm = input(f"Restore backup of {args.cluster} named '{snapshot_name}'? [no] ")
235 if confirm == "yes":
236 print("Restoring cluster.")
237 instances = cluster.get_instances()
238 if len(instances) == 0:
239 print("No instances found. Operation aborted.")
240 else:
241 with log_duration(logger, "restore"):
242 # Stop cluster (not clean)
243 cluster.stop_instances()
244 # Detach and delete current volumes
245 cluster.detach_volumes()
246 cluster.delete_volumes()
247 # Create new volumes from snapshots and attach them
248 cluster.create_volumes(snapshot_name)
249 cluster.attach_volumes(snapshot_name)
250 # Start cluster
251 # cluster.start_instances()
252 print("Operation completed successfully!")
253 else:
254 print("Operation aborted.")
256 if args.list:
257 if args.name is None:
258 print(f"Listing all snapshots associated with {args.cluster}.")
259 snapshots = cluster.get_snapshots("*")
261 # Getting a dictionary of all snapshots associated with the cluster provided grouped by label
262 label_list = []
263 snapshot_dict: dict[str, list[str]] = {}
264 for snapshot in snapshots:
265 for tag in snapshot.tags:
266 if "Label" in tag["Key"] and tag["Value"] not in label_list:
267 label_list.append(tag["Value"])
268 snapshot_dict[tag["Value"]] = []
269 snapshot_dict[tag["Value"]].append(snapshot.id)
270 elif "Label" in tag["Key"]:
271 snapshot_dict[tag["Value"]].append(snapshot.id)
273 # Printing snapshots sorted by Label name
274 label_list.sort()
275 for label in label_list:
276 print("\nLabel: " + label)
277 for snapshot_id in snapshot_dict[label]:
278 print(snapshot_id)
279 else:
280 print(f"Listing snapshots labeled '{args.name}' for {args.cluster}.")
281 snapshots = cluster.get_snapshots(args.name)
282 for snapshot in snapshots:
283 print(snapshot.id)
286if __name__ == "__main__":
287 main()