From e34f7ada9a506e33e9f95e2a139bb7f001766d90 Mon Sep 17 00:00:00 2001 From: sBubshait Date: Tue, 5 Aug 2025 14:27:08 +0300 Subject: [PATCH] feat: allow managing snapshot automations --- snapshots/snapshot.py | 265 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 234 insertions(+), 31 deletions(-) diff --git a/snapshots/snapshot.py b/snapshots/snapshot.py index 50a317e..ca96e1e 100644 --- a/snapshots/snapshot.py +++ b/snapshots/snapshot.py @@ -75,6 +75,13 @@ class PDGSnapshot: if success: for line in output.split('\n'): if self.cron_prefix in line and '#' in line: + # Check if line is paused (commented out) + is_paused = line.strip().startswith('#') and not line.strip().startswith('##') + + if is_paused: + # Remove leading # to parse + line = line.lstrip('#') + parts = line.split('#', 1) if len(parts) == 2: cron_schedule = parts[0].strip() @@ -83,7 +90,7 @@ class PDGSnapshot: dataset = comment.replace(self.cron_prefix, '') automations[dataset] = { 'schedule': cron_schedule, - 'status': 'healthy' + 'status': 'paused' if is_paused else 'active' } return automations @@ -100,6 +107,9 @@ class PDGSnapshot: if dataset not in automations: return "error", "Cron job not found" + if automations[dataset]['status'] == 'paused': + return "paused", "Automation is paused" + return "healthy", "Running normally" def show_loading(self, message: str, duration: float = 2.0): @@ -152,6 +162,99 @@ class PDGSnapshot: except Exception as e: return False, f"Failed to update crontab: {str(e)}" + def pause_cron_job(self, dataset: str) -> Tuple[bool, str]: + """Pause cron job by commenting it out""" + success, current_cron = self.run_command("crontab -l 2>/dev/null") + if not success: + return False, "No crontab found" + + lines = [] + job_found = False + + for line in current_cron.split('\n'): + if line.strip() and self.cron_prefix in line and dataset in line: + if not line.startswith('#'): + lines.append(f"#{line}") # Comment out the line + job_found = True + else: + lines.append(line) # Already paused + job_found = True + else: + lines.append(line) + + if not job_found: + return False, "Automation not found" + + new_cron = '\n'.join(lines) + '\n' + + try: + process = subprocess.Popen(['crontab', '-'], stdin=subprocess.PIPE, text=True) + process.communicate(input=new_cron) + return process.returncode == 0, "Automation paused successfully" + except Exception as e: + return False, f"Failed to pause automation: {str(e)}" + + def resume_cron_job(self, dataset: str) -> Tuple[bool, str]: + """Resume paused cron job by uncommenting it""" + success, current_cron = self.run_command("crontab -l 2>/dev/null") + if not success: + return False, "No crontab found" + + lines = [] + job_found = False + + for line in current_cron.split('\n'): + if line.strip() and self.cron_prefix in line and dataset in line: + if line.startswith('#') and not line.startswith('##'): + lines.append(line[1:]) # Remove comment + job_found = True + else: + lines.append(line) # Already active + job_found = True + else: + lines.append(line) + + if not job_found: + return False, "Automation not found" + + new_cron = '\n'.join(lines) + '\n' + + try: + process = subprocess.Popen(['crontab', '-'], stdin=subprocess.PIPE, text=True) + process.communicate(input=new_cron) + return process.returncode == 0, "Automation resumed successfully" + except Exception as e: + return False, f"Failed to resume automation: {str(e)}" + + def delete_cron_job(self, dataset: str) -> Tuple[bool, str]: + """Delete cron job for dataset""" + success, current_cron = self.run_command("crontab -l 2>/dev/null") + if not success: + return False, "No crontab found" + + lines = [] + job_found = False + + for line in current_cron.split('\n'): + if line.strip() and self.cron_prefix in line and dataset in line: + job_found = True + # Skip this line (delete it) + continue + else: + lines.append(line) + + if not job_found: + return False, "Automation not found" + + new_cron = '\n'.join(lines) + '\n' + + try: + process = subprocess.Popen(['crontab', '-'], stdin=subprocess.PIPE, text=True) + process.communicate(input=new_cron) + return process.returncode == 0, "Automation deleted successfully" + except Exception as e: + return False, f"Failed to delete automation: {str(e)}" + def select_dataset(self) -> Optional[str]: """Interactive dataset selection""" datasets = self.get_zfs_datasets() @@ -236,6 +339,119 @@ class PDGSnapshot: return default_format + def manage_automations(self): + """Manage existing automations (pause/resume/delete)""" + automations = self.get_existing_automations() + + if not automations: + print(f"\n{Colors.YELLOW}No automations found to manage{Colors.ENDC}") + return + + print(f"\n{Colors.BOLD}🔧 Manage Automations{Colors.ENDC}") + datasets = list(automations.keys()) + + # Show automations with status + for i, dataset in enumerate(datasets, 1): + status, _ = self.check_automation_health(dataset) + if status == "healthy": + status_icon = f"{Colors.GREEN}●{Colors.ENDC}" + elif status == "paused": + status_icon = f"{Colors.YELLOW}⏸{Colors.ENDC}" + else: + status_icon = f"{Colors.RED}✗{Colors.ENDC}" + + print(f"{Colors.CYAN}{i:2d}.{Colors.ENDC} {status_icon} {dataset}") + + # Select automation to manage + while True: + try: + choice = input(f"\n{Colors.BOLD}Select automation (1-{len(datasets)}): {Colors.ENDC}") + idx = int(choice) - 1 + if 0 <= idx < len(datasets): + selected_dataset = datasets[idx] + break + print(f"{Colors.RED}Invalid selection{Colors.ENDC}") + except (ValueError, KeyboardInterrupt): + return + + # Show management options + status, _ = self.check_automation_health(selected_dataset) + + print(f"\n{Colors.BOLD}Manage '{selected_dataset}':{Colors.ENDC}") + + options = [] + if status == "healthy": + options.append(("1", "Pause automation", "pause")) + elif status == "paused": + options.append(("1", "Resume automation", "resume")) + + options.append(("2", f"{Colors.RED}Delete automation{Colors.ENDC}", "delete")) + options.append(("3", "Cancel", "cancel")) + + for num, desc, _ in options: + print(f"{Colors.CYAN}{num}.{Colors.ENDC} {desc}") + + # Handle action + while True: + action_choice = input(f"\n{Colors.BOLD}Select action: {Colors.ENDC}") + + action_map = {opt[0]: opt[2] for opt in options} + if action_choice in action_map: + action = action_map[action_choice] + + if action == "cancel": + return + elif action == "pause": + self.show_loading("Pausing automation") + success, message = self.pause_cron_job(selected_dataset) + elif action == "resume": + self.show_loading("Resuming automation") + success, message = self.resume_cron_job(selected_dataset) + elif action == "delete": + confirm = input(f"{Colors.RED}Are you sure? This cannot be undone (y/N): {Colors.ENDC}") + if confirm.lower() != 'y': + return + self.show_loading("Deleting automation") + success, message = self.delete_cron_job(selected_dataset) + + if success: + print(f"{Colors.GREEN}✓ {message}{Colors.ENDC}") + else: + print(f"{Colors.RED}✗ {message}{Colors.ENDC}") + return + + print(f"{Colors.RED}Invalid selection{Colors.ENDC}") + + def show_status(self): + """Display status of all automations""" + automations = self.get_existing_automations() + + if not automations: + print(f"\n{Colors.YELLOW}No automations found{Colors.ENDC}") + return + + print(f"\n{Colors.BOLD}📊 Automation Status{Colors.ENDC}") + + for dataset, info in automations.items(): + status, message = self.check_automation_health(dataset) + + if status == "healthy": + icon = f"{Colors.GREEN}✓{Colors.ENDC}" + status_text = f"{Colors.GREEN}Active{Colors.ENDC}" + elif status == "paused": + icon = f"{Colors.YELLOW}⏸{Colors.ENDC}" + status_text = f"{Colors.YELLOW}Paused{Colors.ENDC}" + else: + icon = f"{Colors.RED}✗{Colors.ENDC}" + status_text = f"{Colors.RED}Error{Colors.ENDC}" + + print(f"\n{icon} {Colors.BOLD}{dataset}{Colors.ENDC}") + print(f" Status: {status_text}") + print(f" Schedule: {Colors.DIM}{info['schedule']}{Colors.ENDC}") + + if status == "error": + print(f" Error: {Colors.RED}{message}{Colors.ENDC}") + def create_automation(self): """Create new snapshot automation""" print(f"\n{Colors.BOLD}📸 Create New Automation{Colors.ENDC}") @@ -270,33 +486,6 @@ class PDGSnapshot: print(f"{Colors.RED}✗ Failed to create automation{Colors.ENDC}") print(f"{Colors.RED}{message}{Colors.ENDC}") - def show_status(self): - """Display status of all automations""" - automations = self.get_existing_automations() - - if not automations: - print(f"\n{Colors.YELLOW}No automations found{Colors.ENDC}") - return - - print(f"\n{Colors.BOLD}📊 Automation Status{Colors.ENDC}") - - for dataset, info in automations.items(): - status, message = self.check_automation_health(dataset) - - if status == "healthy": - icon = f"{Colors.GREEN}✓{Colors.ENDC}" - status_text = f"{Colors.GREEN}Healthy{Colors.ENDC}" - else: - icon = f"{Colors.RED}✗{Colors.ENDC}" - status_text = f"{Colors.RED}Error{Colors.ENDC}" - - print(f"\n{icon} {Colors.BOLD}{dataset}{Colors.ENDC}") - print(f" Status: {status_text}") - print(f" Schedule: {Colors.DIM}{info['schedule']}{Colors.ENDC}") - - if status == "error": - print(f" Error: {Colors.RED}{message}{Colors.ENDC}") - def main_menu(self): """Display main menu and handle user input""" automations = self.get_existing_automations() @@ -305,15 +494,26 @@ class PDGSnapshot: if automation_count > 0: healthy_count = sum(1 for dataset in automations if self.check_automation_health(dataset)[0] == "healthy") - print(f"{Colors.GREEN}📈 {healthy_count}/{automation_count} Automations Healthy{Colors.ENDC}") + paused_count = sum(1 for dataset in automations + if self.check_automation_health(dataset)[0] == "paused") + + status_parts = [] + if healthy_count > 0: + status_parts.append(f"{Colors.GREEN}{healthy_count} Active{Colors.ENDC}") + if paused_count > 0: + status_parts.append(f"{Colors.YELLOW}{paused_count} Paused{Colors.ENDC}") + + status_text = " | ".join(status_parts) if status_parts else f"{Colors.RED}0 Running{Colors.ENDC}" + print(f"{Colors.BOLD}📈 Automations: {status_text}{Colors.ENDC}") print(f"\n{Colors.BOLD}Options:{Colors.ENDC}") print(f"{Colors.CYAN}1.{Colors.ENDC} View Status") print(f"{Colors.CYAN}2.{Colors.ENDC} Create New Automation") - print(f"{Colors.CYAN}3.{Colors.ENDC} Exit") + print(f"{Colors.CYAN}3.{Colors.ENDC} Manage Automations") + print(f"{Colors.CYAN}4.{Colors.ENDC} Exit") while True: - choice = input(f"\n{Colors.BOLD}Select option (1-3): {Colors.ENDC}") + choice = input(f"\n{Colors.BOLD}Select option (1-4): {Colors.ENDC}") if choice == "1": self.show_status() @@ -322,6 +522,9 @@ class PDGSnapshot: self.create_automation() break elif choice == "3": + self.manage_automations() + break + elif choice == "4": print(f"\n{Colors.DIM}Goodbye!{Colors.ENDC}") sys.exit(0) else: