cleaner.py 3.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109
  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 __stale_artifacts_path_list(self) -> list:
  42. """
  43. Returns a list of paths to stale build artifacts.
  44. Returns:
  45. list: A list of file paths for stale artifacts.
  46. """
  47. dir_to_scan = pathlib.Path(bm.get_singleton().get_outdir())
  48. self.logger.debug(
  49. f"Scanning directory: {dir_to_scan} for stale artifacts"
  50. )
  51. all_build_ids = bm.get_singleton().get_all_build_ids()
  52. self.logger.debug(f"Retrieved all build IDs: {all_build_ids}")
  53. dirs_to_keep = [
  54. pathlib.Path(
  55. bm.get_singleton().get_build_artifacts_dir_path(build_id)
  56. )
  57. for build_id in all_build_ids
  58. ]
  59. stale_artifacts = []
  60. for f in dir_to_scan.iterdir():
  61. # Exclude status.json
  62. # To-do: move status.json out of this directory
  63. if f.name == "status.json":
  64. continue
  65. # Check if the current file/dir falls under any directories to keep
  66. keep_file = any(
  67. f.is_relative_to(dir)
  68. for dir in dirs_to_keep
  69. )
  70. if not keep_file:
  71. stale_artifacts.append(str(f))
  72. self.logger.debug(f"Stale artifacts found: {stale_artifacts}")
  73. return stale_artifacts
  74. def __run(self) -> None:
  75. """
  76. Iterates over the list of stale build artifacts
  77. and deletes them from the file system.
  78. """
  79. for path in self.__stale_artifacts_path_list():
  80. self.logger.info(f"Removing stale artifacts at {path}")
  81. shutil.rmtree(path=path)
  82. return
  83. @staticmethod
  84. def get_singleton() -> "BuildArtifactsCleaner":
  85. """
  86. Returns the singleton instance of the BuildArtifactsCleaner class.
  87. Returns:
  88. BuildArtifactsCleaner: The singleton instance of the cleaner.
  89. """
  90. return BuildArtifactsCleaner.__singleton