Merge pull request #23 from SomethingGeneric/copilot/fix-7d16f053-71ca-4b6e-b8e9-a1832bfe7b8e

[WIP] Add per-host snapshot quota in `vm_mapping`

Author: SomethingGeneric

README.md

@@ -176,6 +176,7 @@ Maps Ansible inventory host names to Proxmox VM IDs and nodes:
 ```toml
 # VM Mapping Configuration for miniupdate
 # Maps Ansible inventory host names to Proxmox VM IDs and nodes
+# Optional: Set max_snapshots per VM to limit snapshot count for capacity-limited storage
 
 [vms.web1]
 node = "pve-node1"
@@ -184,12 +185,21 @@ vmid = 100
 [vms.web2] 
 node = "pve-node1"
 vmid = 101
+# Optional: Limit to 2 snapshots for VMs on capacity-limited storage (e.g., small SSDs)
+max_snapshots = 2
 
 [vms.db1]
 node = "pve-node2"
 vmid = 200
 ```
 
+**Per-Host Snapshot Quota:**
+- Use `max_snapshots` to set a maximum number of automated snapshots per VM
+- When set, miniupdate will keep only the N newest snapshots and delete older ones
+- Useful for VMs on capacity-limited storage backends (e.g., small SSDs)
+- Takes precedence over the global `snapshot_retention_days` setting
+- If not set, the global time-based retention policy applies
+
 ### inventory.yml (Ansible Format)
 
 ```yaml

miniupdate/update_automator.py

@@ -486,14 +486,14 @@ def _handle_reboot_and_verification(self, host: Host, vm_mapping: Optional[VMMap
         return None  # Success - no error report needed
 
     def _cleanup_old_snapshots(self, vm_mapping: VMMapping):
-        """Clean up old automated snapshots."""
+        """Clean up old automated snapshots based on retention policy or count limit."""
         try:
-            retention_days = self.update_config.get('snapshot_retention_days', 7)
             prefix = self.update_config.get('snapshot_name_prefix', 'pre-update')
 
             snapshots = self.proxmox_client.list_snapshots(vm_mapping.node, vm_mapping.vmid)
-            cutoff_time = datetime.now() - timedelta(days=retention_days)
 
+            # Filter to only automated snapshots with valid timestamps
+            automated_snapshots = []
             for snapshot in snapshots:
                 snap_name = snapshot.get('name', '')
                 if not snap_name.startswith(prefix):
@@ -503,14 +503,45 @@ def _cleanup_old_snapshots(self, vm_mapping: VMMapping):
                 try:
                     timestamp_str = snap_name[len(prefix) + 1:]  # Remove prefix and dash
                     snap_time = datetime.strptime(timestamp_str, '%Y%m%d-%H%M%S')
-                    
-                    if snap_time < cutoff_time:
-                        logger.info(f"Deleting old snapshot {snap_name} for VM {vm_mapping.vmid}")
-                        self.proxmox_client.delete_snapshot(vm_mapping.node, vm_mapping.vmid, snap_name)
-                        
+                    automated_snapshots.append({
+                        'name': snap_name,
+                        'time': snap_time
+                    })
                 except (ValueError, IndexError) as e:
                     logger.debug(f"Could not parse snapshot timestamp for {snap_name}: {e}")
                     continue
+            
+            # Sort by timestamp, newest first
+            automated_snapshots.sort(key=lambda x: x['time'], reverse=True)
+            
+            snapshots_to_delete = []
+            
+            # Check per-host max_snapshots limit first (takes precedence)
+            if vm_mapping.max_snapshots is not None:
+                # Keep only the newest max_snapshots, delete the rest
+                if len(automated_snapshots) > vm_mapping.max_snapshots:
+                    snapshots_to_delete = automated_snapshots[vm_mapping.max_snapshots:]
+                    logger.info(f"Per-host snapshot quota: keeping {vm_mapping.max_snapshots} newest snapshots "
+                               f"for VM {vm_mapping.vmid}, deleting {len(snapshots_to_delete)} older ones")
+            else:
+                # Fall back to time-based retention policy
+                retention_days = self.update_config.get('snapshot_retention_days', 7)
+                cutoff_time = datetime.now() - timedelta(days=retention_days)
+                
+                snapshots_to_delete = [
+                    snap for snap in automated_snapshots 
+                    if snap['time'] < cutoff_time
+                ]
+                
+                if snapshots_to_delete:
+                    logger.info(f"Time-based retention: deleting {len(snapshots_to_delete)} snapshots "
+                               f"older than {retention_days} days for VM {vm_mapping.vmid}")
+            
+            # Delete the snapshots
+            for snap in snapshots_to_delete:
+                snap_name = snap['name']
+                logger.info(f"Deleting old snapshot {snap_name} for VM {vm_mapping.vmid}")
+                self.proxmox_client.delete_snapshot(vm_mapping.node, vm_mapping.vmid, snap_name)
 
         except Exception as e:
             logger.warning(f"Failed to cleanup old snapshots for VM {vm_mapping.vmid}: {e}")

miniupdate/vm_mapping.py

@@ -18,6 +18,7 @@ class VMMapping(NamedTuple):
     node: str
     vmid: int
     host_name: str
+    max_snapshots: Optional[int] = None
 
 
 class VMMapper:
@@ -72,6 +73,7 @@ def _load_mappings(self) -> Dict[str, VMMapping]:
 
                 node = vm_info.get('node')
                 vmid = vm_info.get('vmid')
+                max_snapshots = vm_info.get('max_snapshots')
 
                 if not node or not vmid:
                     logger.warning(f"Incomplete VM mapping for {host_name}: "
@@ -84,10 +86,22 @@ def _load_mappings(self) -> Dict[str, VMMapping]:
                     logger.warning(f"Invalid vmid for {host_name}: {vmid}")
                     continue
 
+                # Validate max_snapshots if provided
+                if max_snapshots is not None:
+                    try:
+                        max_snapshots = int(max_snapshots)
+                        if max_snapshots < 0:
+                            logger.warning(f"Invalid max_snapshots for {host_name}: must be >= 0")
+                            max_snapshots = None
+                    except ValueError:
+                        logger.warning(f"Invalid max_snapshots for {host_name}: {max_snapshots}")
+                        max_snapshots = None
+                
                 mappings[host_name] = VMMapping(
                     node=node,
                     vmid=vmid,
-                    host_name=host_name
+                    host_name=host_name,
+                    max_snapshots=max_snapshots
                 )
 
             logger.info(f"Loaded VM mappings for {len(mappings)} hosts")
@@ -126,7 +140,8 @@ def create_example_vm_mapping(path: str = "vm_mapping.toml.example") -> None:
             },
             "web2": {
                 "node": "pve-node1", 
-                "vmid": 101
+                "vmid": 101,
+                "max_snapshots": 2  # Optional: limit to 2 snapshots for capacity-limited storage
             },
             "db1": {
                 "node": "pve-node2",
@@ -138,6 +153,7 @@ def create_example_vm_mapping(path: str = "vm_mapping.toml.example") -> None:
     with open(path, 'w', encoding='utf-8') as f:
         # Write with comments
         f.write("# VM Mapping Configuration for miniupdate\n")
-        f.write("# Maps Ansible inventory host names to Proxmox VM IDs and nodes\n\n")
+        f.write("# Maps Ansible inventory host names to Proxmox VM IDs and nodes\n")
+        f.write("# Optional: Set max_snapshots per VM to limit snapshot count for capacity-limited storage\n\n")
 
         toml.dump(example_config, f)