Source code for cleaner

"""
cleaner.py
==========
The cleaner service of mercure. Responsible for deleting processed data after
retention time has passed and if it is offpeak time. Offpeak is the time
period when the cleaning has to be done, because cleaning I/O should be kept
to minimum when receiving and sending exams.
"""

# Standard python includes
import asyncio
import logging
import os
import signal
import sys
import time
from datetime import timedelta, datetime
from datetime import time as _time
from pathlib import Path
from shutil import rmtree, disk_usage
import graphyte
import hupper

# App-specific includes
import common.config as config

import common.helper as helper
import common.monitor as monitor
from common.monitor import task_event
from common.constants import mercure_defs


# Setup daiquiri logger
logger = config.get_logger()

main_loop = None  # type: helper.AsyncTimer # type: ignore


[docs]async def terminate_process(signalNumber, frame) -> None: """Triggers the shutdown of the service.""" helper.g_log("events.shutdown", 1) logger.info("Shutdown requested") monitor.send_event(monitor.m_events.SHUTDOWN_REQUEST, monitor.severity.INFO) # Note: main_loop can be read here because it has been declared as global variable if "main_loop" in globals() and main_loop.is_running: main_loop.stop() helper.trigger_terminate()
[docs]def clean() -> None: """Main entry function.""" if helper.is_terminated(): return helper.g_log("events.run", 1) try: config.read_config() except Exception: logger.warning( # handle_error "Unable to read configuration. Skipping processing.", None, event_type=monitor.m_events.CONFIG_UPDATE, ) return ## Emergency cleaning procedure: Check if server is running out of disk space. If so, clean images right away # Get the percentage of disk usage that should trigger the emergency cleaning emergency_clean_trigger: float = config.mercure.emergency_clean_percentage / 100.0 # Check if the success and discard folder are stored on the same volume success_folder = config.mercure.success_folder discard_folder = config.mercure.discard_folder success_folder_partition = os.stat(success_folder).st_dev discard_folder_partition = os.stat(discard_folder).st_dev # For emergency cleaning need to take into account if success and discard # folders are on the same volume or not. emergency_retention = timedelta(0) folders_to_clear = [success_folder, discard_folder] if success_folder_partition == discard_folder_partition: (total, used, _) = disk_usage(success_folder) bytes_to_clear = int(max(used - total * emergency_clean_trigger, 0)) if bytes_to_clear > 0: for folder in folders_to_clear: # Need to delete all scan data in the both folders to urgently clean up the space. clean_dir(folder, emergency_retention) monitor.send_event( monitor.m_events.PROCESSING, monitor.severity.WARNING, f"Disk is almost full. Emergency cleaning of the {success_folder} and {discard_folder} folders. Consider adjusting retention period.", ) else: bytes_to_clear = 0 for folder in folders_to_clear: (total, used, _) = disk_usage(folder) bytes_to_clear = int(max(used - total * emergency_clean_trigger, 0)) if bytes_to_clear > 0: # Need to delete all scan data in the folder to urgently clean up the space. clean_dir(folder, emergency_retention) monitor.send_event( monitor.m_events.PROCESSING, monitor.severity.WARNING, f"Disk is almost full. Emergency cleaning of the {folder} folder. Consider adjusting retention period.", ) ## Regular cleaning procedure if _is_offpeak( config.mercure.offpeak_start, config.mercure.offpeak_end, datetime.now().time(), ): retention = timedelta(seconds=config.mercure.retention) clean_dir(success_folder, retention) clean_dir(discard_folder, retention)
def _is_offpeak(offpeak_start: str, offpeak_end: str, current_time: _time) -> bool: try: start_time = datetime.strptime(offpeak_start, "%H:%M").time() end_time = datetime.strptime(offpeak_end, "%H:%M").time() except Exception as e: logger.error(f"Unable to parse offpeak time: {offpeak_start}, {offpeak_end}", None) # handle_error return True if start_time < end_time: return current_time >= start_time and current_time <= end_time # End time is after midnight return current_time >= start_time or current_time <= end_time
[docs]def clean_dir(folder, retention) -> None: """ Cleans items from the given folder that have exceeded the retention time, starting with the oldest items """ candidates = [ (f, f.stat().st_mtime) for f in Path(folder).iterdir() if f.is_dir() and retention < timedelta(seconds=(time.time() - f.stat().st_mtime)) ] for entry in candidates: delete_folder(entry)
[docs]def delete_folder(entry) -> None: """Deletes given folder.""" delete_path = entry[0] series_uid = find_series_uid(delete_path) try: rmtree(delete_path) logger.info(f"Deleted folder {delete_path} from {series_uid}") monitor.send_task_event(task_event.CLEAN, Path(delete_path).stem, 0, delete_path, "Deleted folder") except Exception as e: logger.error( f"Unable to delete folder {delete_path}", Path(delete_path).stem, target=delete_path, ) # handle_error
[docs]def find_series_uid(work_dir) -> str: """ Finds series uid which is always part before the '#'-sign in filename. """ to_be_deleted_dir = Path(work_dir) for entry in to_be_deleted_dir.iterdir(): if "#" in entry.name: return entry.name.split(mercure_defs.SEPARATOR)[0] return "series_uid-not-found" return "series_uid-not-found"
[docs]def exit_cleaner(args) -> None: """Stop the asyncio event loop.""" helper.loop.call_soon_threadsafe(helper.loop.stop)
[docs]def main(args=sys.argv[1:]) -> None: if "--reload" in args or os.getenv("MERCURE_ENV", "PROD").lower() == "dev": # start_reloader will only return in a monitored subprocess reloader = hupper.start_reloader("cleaner.main") import logging logging.getLogger("watchdog").setLevel(logging.WARNING) logger.info("") logger.info(f"mercure DICOM Cleaner ver {mercure_defs.VERSION}") logger.info("--------------------------------------------") logger.info("") # Register system signals to be caught signals = (signal.SIGTERM, signal.SIGINT) for s in signals: helper.loop.add_signal_handler(s, lambda s=s: asyncio.create_task(terminate_process(s, helper.loop))) instance_name = "main" if len(sys.argv) > 1: instance_name = sys.argv[1] try: config.read_config() except Exception: logger.exception("Cannot start service. Going down.") sys.exit(1) appliance_name = config.mercure.appliance_name logger.info(f"Appliance name = {appliance_name}") logger.info(f"Instance name = {instance_name}") logger.info(f"Instance PID = {os.getpid()}") logger.info(sys.version) monitor.configure("cleaner", instance_name, config.mercure.bookkeeper) monitor.send_event(monitor.m_events.BOOT, monitor.severity.INFO, f"PID = {os.getpid()}") if len(config.mercure.graphite_ip) > 0: logger.info(f"Sending events to graphite server: {config.mercure.graphite_ip}") graphite_prefix = "mercure." + appliance_name + ".cleaner." + instance_name graphyte.init( config.mercure.graphite_ip, config.mercure.graphite_port, prefix=graphite_prefix, ) global main_loop main_loop = helper.AsyncTimer(config.mercure.cleaner_scan_interval, clean) main_loop.start() helper.g_log("events.boot", 1) try: # Start the asyncio event loop for asynchronous function calls main_loop.run_until_complete(helper.loop) # Process will exit here once the asyncio loop has been stopped monitor.send_event(monitor.m_events.SHUTDOWN, monitor.severity.INFO) except Exception as e: # Process will exit here once the asyncio loop has been stopped monitor.send_event(monitor.m_events.SHUTDOWN, monitor.severity.ERROR, str(e)) finally: # Finish all asyncio tasks that might be still pending remaining_tasks = helper.asyncio.all_tasks(helper.loop) # type: ignore[attr-defined] if remaining_tasks: helper.loop.run_until_complete(helper.asyncio.gather(*remaining_tasks)) logger.info("Going down now")
if __name__ == "__main__": main()