cleaner.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111
  1. from utils import TaskRunner
  2. from .manager import BuildManager as bm
  3. import logging
  4. import shutil
  5. import pathlib
  6. class BuildArtifactsCleaner:
  7. """
  8. Class responsible for cleaning up stale build
  9. artifacts from the build output directory.
  10. """
  11. __singleton = None
  12. def __init__(self) -> None:
  13. """
  14. Initialises the BuildArtifactsCleaner instance.
  15. This class depends on the BuildManager singleton,
  16. so it ensures that the BuildManager is initialized
  17. before proceeding.
  18. Raises:
  19. RuntimeError: If BuildManager is not initialized or
  20. if another instance of BuildArtifactsCleaner already
  21. exists (enforcing the singleton pattern).
  22. """
  23. if bm.get_singleton() is None:
  24. raise RuntimeError("BuildManager should be initialised first")
  25. if BuildArtifactsCleaner.__singleton:
  26. raise RuntimeError("BuildArtifactsCleaner must be a singleton")
  27. # Calls the __run method every 60 seconds.
  28. tasks = (
  29. (self.__run, 60),
  30. )
  31. # This spins up a new thread
  32. self.__runner = TaskRunner(tasks=tasks)
  33. self.logger = logging.getLogger(__name__)
  34. BuildArtifactsCleaner.__singleton = self
  35. def start(self) -> None:
  36. """
  37. Start BuildArtifactsCleaner.
  38. """
  39. self.logger.info("Starting BuildArtifactsCleaner")
  40. self.__runner.start()
  41. def stop(self) -> None:
  42. """
  43. Stop BuildArtifactsCleaner.
  44. """
  45. self.logger.info("Stopping BuildArtifactsCleaner")
  46. self.__runner.stop()
  47. def __stale_artifacts_path_list(self) -> list:
  48. """
  49. Returns a list of paths to stale build artifacts.
  50. Returns:
  51. list: A list of file paths for stale artifacts.
  52. """
  53. dir_to_scan = pathlib.Path(bm.get_singleton().get_outdir())
  54. self.logger.debug(
  55. f"Scanning directory: {dir_to_scan} for stale artifacts"
  56. )
  57. all_build_ids = bm.get_singleton().get_all_build_ids()
  58. self.logger.debug(f"Retrieved all build IDs: {all_build_ids}")
  59. dirs_to_keep = [
  60. pathlib.Path(
  61. bm.get_singleton().get_build_artifacts_dir_path(build_id)
  62. )
  63. for build_id in all_build_ids
  64. ]
  65. stale_artifacts = []
  66. for f in dir_to_scan.iterdir():
  67. # Check if the current file/dir falls under any directories to keep
  68. keep_file = any(
  69. f.is_relative_to(dir)
  70. for dir in dirs_to_keep
  71. )
  72. if not keep_file:
  73. stale_artifacts.append(str(f))
  74. self.logger.debug(f"Stale artifacts found: {stale_artifacts}")
  75. return stale_artifacts
  76. def __run(self) -> None:
  77. """
  78. Iterates over the list of stale build artifacts
  79. and deletes them from the file system.
  80. """
  81. for path in self.__stale_artifacts_path_list():
  82. self.logger.info(f"Removing stale artifacts at {path}")
  83. shutil.rmtree(path=path)
  84. return
  85. @staticmethod
  86. def get_singleton() -> "BuildArtifactsCleaner":
  87. """
  88. Returns the singleton instance of the BuildArtifactsCleaner class.
  89. Returns:
  90. BuildArtifactsCleaner: The singleton instance of the cleaner.
  91. """
  92. return BuildArtifactsCleaner.__singleton