diff --git a/roles/system-maintenance-service-freezer/files/system-maintenance-service-freezer.py b/roles/system-maintenance-service-freezer/files/system-maintenance-service-freezer.py index 2a207d0f..12a510db 100644 --- a/roles/system-maintenance-service-freezer/files/system-maintenance-service-freezer.py +++ b/roles/system-maintenance-service-freezer/files/system-maintenance-service-freezer.py @@ -2,12 +2,18 @@ import argparse import subprocess import time import os +from datetime import datetime def parse_time_to_seconds(time_str): """Convert time string to seconds.""" units = {"s": 1, "min": 60, "h": 3600} - number, unit = time_str[:-1], time_str[-1] - if unit not in units: + if time_str[-3:] in units: + number, unit = time_str[:-3], time_str[-3:] + elif time_str[-2:] in units: + number, unit = time_str[:-2], time_str[-2:] + elif time_str[-1:] in units: + number, unit = time_str[:-1], time_str[-1:] + else: raise ValueError("Invalid time unit") return int(number) * units[unit] @@ -30,78 +36,89 @@ def check_service_active(service_name): service_status = result.stdout.decode('utf-8').strip() return_value=service_status in ['active', 'activating'] if return_value: - print(f"Service {service} is active."); + print(f"Service {service_name} is active."); else: - print(f"Service {service} is not active."); + print(f"Service {service_name} is not active."); return return_value def check_any_service_active(services): """Check if any service in the list is active or activating.""" - for service in services: + for service_name in services: if check_service_active(service_name): return True return False + +def stop_timer(service): + # Stop and disable the corresponding timer, if it exists + if service_file_exists(service,"timer"): + timer_name = service + ".timer" + subprocess.run(['systemctl', 'stop', timer_name]) + subprocess.run(['systemctl', 'disable', timer_name]) + print(f"{timer_name} stopped and disabled.") + else: + print(f"No timer to stop.") + +def filter_services(services, ignored_services): + """ + Removes all services from the 'services' list that are contained in 'ignored_services'. -def freeze(services_to_wait_for, ignored_services, timeout_sec): + :param services: List of services. + :param ignored_services: List of services to be ignored. + :return: Filtered list of services. + """ + return [service for service in services if service not in ignored_services] + +def stop_all_timers(services): + for service in services: + stop_timer(service) + +def wait_for_all_services_to_stop(filtered_services,max_attempts,attempt,break_time_sec): + for service in filtered_services: + while check_service_active(service): + attempt += 1 + if attempt > max_attempts: + raise Exception(f"Error: Maximum attempts ({max_attempts}) reached. Exit.") + print(f"{datetime.now().isoformat()}#{attempt}/{max_attempts}: Waiting for {break_time_sec} seconds for {service} to stop...") + time.sleep(break_time_sec) + return attempt + +def freeze(filtered_services,timeout_sec): break_time_sec=5 attempt=0 max_attempts=timeout_sec/break_time_sec - any_service_active=True - print("Checking if any services is active.") # This guaranties that no service will be started in the between time - while any_service_active: - # Filter services that exist and are not in the ignored list - for service in services_to_wait_for: - print(f"\nFreezing: {service}") - if service in ignored_services: - print(f"{service} will be ignored.") - else: - while check_service_active(service): - current_time_iso = datetime.now().isoformat() - attempt += 1 - print(f"{current_time_iso}#{attempt}/{max_attempts}: Waiting for {break_time_sec} seconds for {service} to stop...") - time.sleep(break_time_sec) - if attempt > max_attempts: - raise Exception(f"Error: Maximum attempts ({max_attempts}) reached. Exit.") - - # Stop and disable the corresponding timer, if it exists - if service_file_exists(service,"timer"): - timer_name = service + ".timer" - subprocess.run(['systemctl', 'stop', timer_name]) - subprocess.run(['systemctl', 'disable', timer_name]) - print(f"{timer_name} stopped and disabled.") - else: - print(f"No timer to stop.") - print("Checking if any of the previous deactivated services is active.") - any_service_active=check_any_service_active(services_to_wait_for) + while check_any_service_active(filtered_services): + stop_all_timers(filtered_services) + attempt=wait_for_all_services_to_stop(filtered_services,max_attempts,attempt,break_time_sec) print("\nAll required services have stopped.") -def defrost(services_to_wait_for, ignored_services): - for service in services_to_wait_for: +def defrost(filtered_services): + for service in filtered_services: print(f"\nUnfreezing: {service}") - if service in ignored_services: - print(f"{service} will be ignored.") - elif service_file_exists(service,"timer"): + if service_file_exists(service,"timer"): # Start and enable the corresponding timer, if it exists timer_name = service + ".timer" subprocess.run(['systemctl', 'start', timer_name]) subprocess.run(['systemctl', 'enable', timer_name]) print(f"{timer_name} started and enabled.") else: - print(f"Skipped.") + print(f"No timer to activate.") print("\nAll required services are started.") -def main(services_to_wait_for, ignored_services, action, timeout_sec): - print(f"Services to wait for: {services_to_wait_for}") - print(f"Services to ignore: {ignored_services}") +def main(services, ignored_services, action, timeout_sec): + filtered_services=filter_services(services, ignored_services) + print(f"Services to handle : {services}") + print(f"Services to ignore : {ignored_services}") + print(f"Services filtered : {filtered_services}") + if action == 'freeze': print("Freezing services."); - freeze(services_to_wait_for, ignored_services, timeout_sec) + freeze(filtered_services, timeout_sec) elif action == 'defrost': print("Unfreezing services."); - defrost(services_to_wait_for, ignored_services) + defrost(filtered_services) print('\nOverview:') subprocess.run(['systemctl', 'list-timers']) @@ -111,9 +128,8 @@ if __name__ == "__main__": parser.add_argument('services', nargs='+', help='List of services to apply the action to') parser.add_argument('--ignore', nargs='*', help='List of services to ignore in the action', default=[]) parser.add_argument('--timeout', help='Timeout for freezing services (e.g., 1h, 30min, 45s)', default='1h') - args = parser.parse_args() - services_to_wait_for = args.services + services = args.services ignored_services = args.ignore if args.ignore else [] timeout_seconds = parse_time_to_seconds(args.timeout) - main(services_to_wait_for, ignored_services, args.action, timeout_seconds) + main(services, ignored_services, args.action, timeout_seconds)