build_manager: initial implementation of the module

build_manager/__init__.py

@@ -0,0 +1,17 @@
+from .cleaner import BuildArtifactsCleaner
+from .progress_updater import BuildProgressUpdater
+from .manager import (
+    BuildManager,
+    BuildInfo,
+    BuildProgress,
+    BuildState,
+)
+
+__all__ = [
+    "BuildArtifactsCleaner",
+    "BuildProgressUpdater",
+    "BuildManager",
+    "BuildInfo",
+    "BuildProgress",
+    "BuildState",
+]

build_manager/cleaner.py

@@ -0,0 +1,109 @@
+from utils import TaskRunner
+from .manager import BuildManager as bm
+import logging
+import shutil
+import pathlib
+
+
+class BuildArtifactsCleaner:
+    """
+    Class responsible for cleaning up stale build
+    artifacts from the build output directory.
+    """
+
+    __singleton = None
+
+    def __init__(self) -> None:
+        """
+        Initialises the BuildArtifactsCleaner instance.
+        This class depends on the BuildManager singleton,
+        so it ensures that the BuildManager is initialized
+        before proceeding.
+
+        Raises:
+            RuntimeError: If BuildManager is not initialized or
+            if another instance of BuildArtifactsCleaner already
+            exists (enforcing the singleton pattern).
+        """
+
+        if bm.get_singleton() is None:
+            raise RuntimeError("BuildManager should be initialised first")
+
+        if BuildArtifactsCleaner.__singleton:
+            raise RuntimeError("BuildArtifactsCleaner must be a singleton")
+
+        # Calls the __run method every 60 seconds.
+        tasks = (
+            (self.__run, 60),
+        )
+        # This spins up a new thread
+        self.__runner = TaskRunner(tasks=tasks)
+        self.logger = logging.getLogger(__name__)
+        BuildArtifactsCleaner.__singleton = self
+
+    def start(self) -> None:
+        """
+        Start BuildArtifactsCleaner.
+        """
+        self.logger.info("Starting BuildArtifactsCleaner")
+        self.__runner.start()
+
+    def __stale_artifacts_path_list(self) -> list:
+        """
+        Returns a list of paths to stale build artifacts.
+
+        Returns:
+            list: A list of file paths for stale artifacts.
+        """
+        dir_to_scan = pathlib.Path(bm.get_singleton().get_outdir())
+        self.logger.debug(
+            f"Scanning directory: {dir_to_scan} for stale artifacts"
+        )
+        all_build_ids = bm.get_singleton().get_all_build_ids()
+        self.logger.debug(f"Retrieved all build IDs: {all_build_ids}")
+
+        dirs_to_keep = [
+            pathlib.Path(
+                bm.get_singleton().get_build_artifacts_dir_path(build_id)
+            )
+            for build_id in all_build_ids
+        ]
+
+        stale_artifacts = []
+        for f in dir_to_scan.iterdir():
+            # Exclude status.json
+            # To-do: move status.json out of this directory
+            if f.name == "status.json":
+                continue
+
+            # Check if the current file/dir falls under any directories to keep
+            keep_file = any(
+                f.is_relative_to(dir)
+                for dir in dirs_to_keep
+            )
+            if not keep_file:
+                stale_artifacts.append(str(f))
+
+        self.logger.debug(f"Stale artifacts found: {stale_artifacts}")
+
+        return stale_artifacts
+
+    def __run(self) -> None:
+        """
+        Iterates over the list of stale build artifacts
+        and deletes them from the file system.
+        """
+        for path in self.__stale_artifacts_path_list():
+            self.logger.info(f"Removing stale artifacts at {path}")
+            shutil.rmtree(path=path)
+        return
+
+    @staticmethod
+    def get_singleton() -> "BuildArtifactsCleaner":
+        """
+        Returns the singleton instance of the BuildArtifactsCleaner class.
+
+        Returns:
+            BuildArtifactsCleaner: The singleton instance of the cleaner.
+        """
+        return BuildArtifactsCleaner.__singleton

build_manager/manager.py

@@ -0,0 +1,445 @@
+import time
+import redis
+import dill
+from enum import Enum
+from utils import RateLimiter
+import logging
+import hashlib
+from metadata_manager import RemoteInfo
+import os
+
+
+class BuildState(Enum):
+    PENDING = 0
+    RUNNING = 1
+    SUCCESS = 2
+    FAILURE = 3
+    ERROR = 4
+
+
+class BuildProgress:
+    def __init__(
+        self,
+        state: BuildState,
+        percent: int
+    ) -> None:
+        """
+        Initialise the progress property for a build,
+        including its state and completion percentage.
+
+        Parameters:
+            state (BuildState): The current state of the build.
+            percent (int): The completion percentage of the build (0-100).
+        """
+        self.state = state
+        self.percent = percent
+
+
+class BuildInfo:
+    def __init__(self,
+                 vehicle: str,
+                 remote_info: RemoteInfo,
+                 git_hash: str,
+                 board: str,
+                 selected_features: set) -> None:
+        """
+        Initialize build information object including vehicle,
+        remote, git hash, selected features, and progress of the build.
+        The progress percentage is initially 0 and the state is PENDING.
+
+        Parameters:
+            vehicle (str): The vehicle name or type associated with the build.
+            remote_info (RemoteInfo): The remote repository containing the
+            source commit to build on.
+            git_hash (str): The git commit hash to build on.
+            board (str): Board to build for.
+            selected_features (set): Set of features selected for the build.
+        """
+        self.vehicle = vehicle
+        self.remote_info = remote_info
+        self.git_hash = git_hash
+        self.board = board
+        self.selected_features = selected_features
+        self.progress = BuildProgress(
+            state=BuildState.PENDING,
+            percent=0
+        )
+        self.time_created = time.time()
+
+
+class BuildManager:
+    """
+    Class to manage the build lifecycle, including build submission,
+    announcements, progress updates, and retrieval of build-related
+    information.
+    """
+
+    __singleton = None
+
+    def __init__(self,
+                 outdir: str,
+                 redis_host: str = 'localhost',
+                 redis_port: int = 6379,
+                 redis_task_queue_name: str = 'builds-queue') -> None:
+        """
+        Initialide the BuildManager instance. This class is responsible
+        for interacting with Redis to store build metadata and managing
+        build tasks.
+
+        Parameters:
+            outdir (str): Path to the directory for storing build artifacts.
+            redis_host (str): Hostname of the Redis instance for storing build
+            metadata.
+            redis_port (int): Port of the Redis instance for storing build
+            metadata.
+            redis_task_queue_name (str): Redis List name to be used as the
+            task queue.
+
+        Raises:
+            RuntimeError: If an instance of this class already exists,
+            enforcing a singleton pattern.
+        """
+        if BuildManager.__singleton:
+            raise RuntimeError("BuildManager must be a singleton")
+
+        # Initialide Redis client without decoding responses
+        # as we use dill for serialization.
+        self.__redis_client = redis.Redis(
+            host=redis_host,
+            port=redis_port,
+            decode_responses=False
+        )
+        self.__task_queue = redis_task_queue_name
+        self.__outdir = outdir
+
+        # Initialide an IP-based rate limiter.
+        # Allow 10 builds per hour per client
+        self.__ip_rate_limiter = RateLimiter(
+            redis_host=redis_host,
+            redis_port=redis_port,
+            time_window_sec=3600,
+            allowed_requests=10
+        )
+        self.__build_entry_prefix = "buildmeta-"
+        self.logger = logging.getLogger(__name__)
+        self.logger.info(
+            "Build Manager initialised with configuration: "
+            f"Redis host: {redis_host}, "
+            f"Redis port: {redis_port}, "
+            f"Redis task queue: {self.__task_queue}, "
+            f"Build output directory: {self.__outdir}, "
+            f"Build entry prefix: {self.__build_entry_prefix}"
+        )
+        BuildManager.__singleton = self
+
+    def __del__(self) -> None:
+        """
+        Gracefully close the Redis connection when the BuildManager instance
+        is deleted.
+        """
+        if self.__redis_client:
+            self.logger.debug("Closing Redis connection")
+            self.__redis_client.close()
+
+    def __key_from_build_id(self, build_id: str) -> str:
+        """
+        Generate the Redis key that stores the build information for the given
+        build ID.
+
+        Parameters:
+            build_id (str): The unique ID for the build.
+
+        Returns:
+            str: The Redis key containing the build information.
+        """
+        return self.__build_entry_prefix + build_id
+
+    def __build_id_from_key(self, key: str) -> str:
+        """
+        Extract the build ID from the given Redis key.
+
+        Parameters:
+            key (str): The Redis key storing build information.
+
+        Returns:
+            str: The build ID corresponding to the given Redis key.
+        """
+        return key[len(self.__build_entry_prefix):]
+
+    def get_outdir(self) -> str:
+        """
+        Return the directory where build artifacts are stored.
+
+        Returns:
+            str: Path to the output directory containing build artifacts.
+        """
+        return self.__outdir
+
+    def __generate_build_id(self, build_info: BuildInfo) -> str:
+        """
+        Generate a unique build ID based on the build information and
+        current timestamp. The build information is hashed and combined
+        with the time to generate the ID.
+
+        Parameters:
+            build_info (BuildInfo): The build information object.
+
+        Returns:
+            str: The generated build ID (64 characters).
+        """
+        h = hashlib.md5(
+            f"{build_info}-{time.time_ns()}".encode()
+        ).hexdigest()
+        bid = f"{build_info.vehicle}-{build_info.board}-{h}"
+        return bid
+
+    def submit_build(self,
+                     build_info: BuildInfo,
+                     client_ip: str) -> str:
+        """
+        Submit a new build request, generate a build ID, and queue the
+        build for processing.
+
+        Parameters:
+            build_info (BuildInfo): The build information.
+            client_ip (str): The IP address of the client submitting the
+            build request.
+
+        Returns:
+            str: The generated build ID for the submitted build.
+        """
+        self.__ip_rate_limiter.count(client_ip)
+        build_id = self.__generate_build_id(build_info)
+        self.__insert_build_info(build_id=build_id, build_info=build_info)
+        self.__queue_build(build_id=build_id)
+        return build_id
+
+    def __queue_build(self,
+                      build_id: str) -> None:
+        """
+        Add the build ID to the Redis task queue for processing.
+
+        Parameters:
+            build_id (str): The ID of the build to be queued.
+        """
+        self.__redis_client.rpush(
+            self.__task_queue,
+            build_id.encode()
+        )
+
+    def get_next_build_id(self) -> str:
+        """
+        Block until the next build ID is available in the task queue,
+        then return it.
+
+        Returns:
+            str: The ID of the next build to be processed.
+        """
+        _, build_id_encoded = self.__redis_client.blpop(self.__task_queue)
+        build_id = build_id_encoded.decode()
+        self.logger.debug(f"Next build id: {build_id}")
+        return build_id
+
+    def build_exists(self,
+                     build_id: str) -> bool:
+        """
+        Check if a build with the given ID exists in the datastore.
+
+        Parameters:
+            build_id (str): The ID of the build to check.
+
+        Returns:
+            bool: True if the build exists, False otherwise.
+        """
+        return self.__redis_client.exists(
+            self.__key_from_build_id(build_id=build_id)
+        )
+
+    def __insert_build_info(self,
+                            build_id: str,
+                            build_info: BuildInfo,
+                            ttl_sec: int = 86400) -> None:
+        """
+        Insert the build information into the datastore.
+
+        Parameters:
+            build_id (str): The ID of the build.
+            build_info (BuildInfo): The build information to store.
+            ttl_sec (int): Time-to-live (TTL) in seconds after which the
+            build expires.
+        """
+        if self.build_exists(build_id=build_id):
+            raise ValueError(f"Build with id {build_id} already exists")
+
+        key = self.__key_from_build_id(build_id)
+        self.logger.debug(
+            "Adding build info, "
+            f"Redis key: {key}, "
+            f"Build Info: {build_info}, "
+            f"TTL: {ttl_sec} sec"
+        )
+        self.__redis_client.set(
+            name=key,
+            value=dill.dumps(build_info),
+            ex=ttl_sec
+        )
+
+    def get_build_info(self,
+                       build_id: str) -> BuildInfo:
+        """
+        Retrieve the build information for the given build ID.
+
+        Parameters:
+            build_id (str): The ID of the build to retrieve.
+
+        Returns:
+            BuildInfo: The build information for the given build ID.
+        """
+        key = self.__key_from_build_id(build_id=build_id)
+        self.logger.debug(
+            f"Getting build info for build id {build_id}, Redis Key: {key}"
+        )
+        value = self.__redis_client.get(key)
+        self.logger.debug(f"Got value {value} at key {key}")
+        return dill.loads(value) if value else None
+
+    def __update_build_info(self,
+                            build_id: str,
+                            build_info: BuildInfo) -> None:
+        """
+        Update the build information for an existing build in datastore.
+
+        Parameters:
+            build_id (str): The ID of the build to update.
+            build_info (BuildInfo): The new build information to replace
+            the existing one.
+        """
+        key = self.__key_from_build_id(build_id=build_id)
+        self.logger.debug(
+            "Updating build info, "
+            f"Redis key: {key}, "
+            f"Build Info: {build_info}, "
+            f"TTL: Keeping Same"
+        )
+        self.__redis_client.set(
+            name=key,
+            value=dill.dumps(build_info),
+            keepttl=True
+        )
+
+    def update_build_progress_percent(self,
+                                      build_id: str,
+                                      percent: int) -> None:
+        """
+        Update the build's completion percentage.
+
+        Parameters:
+            build_id (str): The ID of the build to update.
+            percent (int): The new completion percentage (0-100).
+        """
+        build_info = self.get_build_info(build_id=build_id)
+
+        if build_info is None:
+            raise ValueError(f"Build with id {build_id} not found.")
+
+        build_info.progress.percent = percent
+        self.__update_build_info(
+            build_id=build_id,
+            build_info=build_info
+        )
+
+    def update_build_progress_state(self,
+                                    build_id: str,
+                                    new_state: BuildState) -> None:
+        """
+        Update the build's state (e.g., PENDING, RUNNING, SUCCESS, FAILURE).
+
+        Parameters:
+            build_id (str): The ID of the build to update.
+            new_state (BuildState): The new state to set for the build.
+        """
+        build_info = self.get_build_info(build_id=build_id)
+
+        if build_info is None:
+            raise ValueError(f"Build with id {build_id} not found.")
+
+        build_info.progress.state = new_state
+        self.__update_build_info(
+            build_id=build_id,
+            build_info=build_info
+        )
+
+    def get_all_build_ids(self) -> list:
+        """
+        Retrieve the IDs of all builds currently stored in the datastore.
+
+        Returns:
+            list: A list of all build IDs.
+        """
+        keys_encoded = self.__redis_client.keys(
+            f"{self.__build_entry_prefix}*"
+        )
+        keys = [key.decode() for key in keys_encoded]
+        self.logger.debug(
+            f"Keys with prefix {self.__build_entry_prefix}"
+            f": {keys}"
+        )
+        return [
+            self.__build_id_from_key(key)
+            for key in keys
+        ]
+
+    def get_build_artifacts_dir_path(self, build_id: str) -> str:
+        """
+        Return the directory at which the build artifacts are stored.
+
+        Parameters:
+            build_id (str): The ID of the build.
+
+        Returns:
+            str: The build artifacts path.
+        """
+        return os.path.join(
+            self.get_outdir(),
+            build_id,
+        )
+
+    def get_build_log_path(self, build_id: str) -> str:
+        """
+        Return the path at which the log for a build is written.
+
+        Parameters:
+            build_id (str): The ID of the build.
+
+        Returns:
+            str: The path at which the build log is written.
+        """
+        return os.path.join(
+            self.get_build_artifacts_dir_path(build_id),
+            'build.log'
+        )
+
+    def get_build_archive_path(self, build_id: str) -> str:
+        """
+        Return the path to the build archive.
+
+        Parameters:
+            build_id (str): The ID of the build.
+
+        Returns:
+            str: The path to the build archive.
+        """
+        return os.path.join(
+            self.get_build_artifacts_dir_path(build_id),
+            f"{build_id}.tar.gz"
+        )
+
+    @staticmethod
+    def get_singleton() -> "BuildManager":
+        """
+        Return the singleton instance of the BuildManager class.
+
+        Returns:
+            BuildManager: The singleton instance of the BuildManager.
+        """
+        return BuildManager.__singleton

build_manager/progress_updater.py

@@ -0,0 +1,353 @@
+import re
+import os
+import logging
+from utils import TaskRunner
+from pathlib import Path
+from .manager import (
+    BuildManager as bm,
+    BuildState
+)
+import json
+import time
+
+
+class BuildProgressUpdater:
+    """
+    Class for updating the progress of all builds.
+
+    This class ensures that the progress of all  builds is
+    updated periodically. It operates in a singleton pattern
+    to ensure only one instance manages the updates.
+    """
+
+    __singleton = None
+
+    def __init__(self):
+        """
+        Initialises the BuildProgressUpdater instance.
+
+        This uses the BuildManager singleton, so ensure that BuildManager is
+        initialised before creating a BuildProgressUpdater instance.
+
+        Raises:
+            RuntimeError: If BuildManager is not initialized or
+            if another instance of BuildProgressUpdater has already
+            been initialised.
+        """
+        if not bm.get_singleton():
+            raise RuntimeError("BuildManager should be initialised first")
+
+        if BuildProgressUpdater.__singleton:
+            raise RuntimeError("BuildProgressUpdater must be a singleton.")
+
+        self.__ensure_status_json()
+        # Set up a periodic task to update build progress every 3 seconds
+        # TaskRunner will handle scheduling and running the task.
+        tasks = (
+            (self.__update_build_progress_all, 3),
+        )
+        self.__runner = TaskRunner(tasks=tasks)
+        self.logger = logging.getLogger(__name__)
+        BuildProgressUpdater.__singleton = self
+
+    def start(self) -> None:
+        """
+        Start BuildProgressUpdater.
+        """
+        self.logger.info("Starting BuildProgressUpdater.")
+        self.__runner.start()
+
+    def __calc_running_build_progress_percent(self, build_id: str) -> int:
+        """
+        Calculate the progress percentage of a running build.
+
+        This method analyses the build log to determine the current completion
+        percentage by parsing the build steps from the log file.
+
+        Parameters:
+            build_id (str): The unique ID of the build for which progress is
+            calculated.
+
+        Returns:
+            int: The calculated build progress percentage (0 to 100).
+
+        Raises:
+            ValueError: If no build information is found for the provided
+            build ID.
+        """
+        build_info = bm.get_singleton().get_build_info(build_id=build_id)
+
+        if build_info is None:
+            raise ValueError(f"No build found with ID {build_id}")
+
+        if build_info.progress.state != BuildState.RUNNING:
+            raise RuntimeError(
+                "This method should only be called for running builds."
+            )
+
+        # Construct path to the build's log file
+        log_file_path = bm.get_singleton().get_build_log_path(build_id)
+        self.logger.debug(f"Opening log file: {log_file_path}")
+
+        try:
+            # Read the log content
+            with open(log_file_path, encoding='utf-8') as f:
+                build_log = f.read()
+        except FileNotFoundError:
+            self.logger.error(
+                f"Log file not found for RUNNING build with ID: {build_id}"
+            )
+            return build_info.progress.percent
+
+        # Regular expression to extract the build progress steps
+        compiled_regex = re.compile(r'(\[\D*(\d+)\D*\/\D*(\d+)\D*\])')
+        self.logger.debug(f"Regex pattern: {compiled_regex}")
+        all_matches = compiled_regex.findall(build_log)
+        self.logger.debug(f"Log matches: {all_matches}")
+
+        # If no matches are found, return a default progress value of 0
+        if len(all_matches) < 1:
+            return 0
+
+        completed_steps, total_steps = all_matches[-1][1:]
+        self.logger.debug(
+            f"Completed steps: {completed_steps},"
+            f"Total steps: {total_steps}"
+        )
+
+        # Handle initial compilation/linking steps (minor weight)
+        if int(total_steps) < 20:
+            return 1
+
+        # Handle building the OS phase (4% weight)
+        if int(total_steps) < 200:
+            return (int(completed_steps) * 4 // int(total_steps)) + 1
+
+        # Major build phase (95% weight)
+        return (int(completed_steps) * 95 // int(total_steps)) + 5
+
+    def __refresh_running_build_state(self, build_id: str) -> BuildState:
+        """
+        Refresh the state of a running build.
+
+        This method analyses the build log to determine the build has
+        concluded. If yes, it detects the success of a build by finding
+        the success message in the log.
+
+        Parameters:
+            build_id (str): The unique ID of the build for which progress is
+            calculated.
+
+        Returns:
+            BuildSate: The current build state based on the log.
+
+        Raises:
+            ValueError: If no build information is found for the provided
+            build ID.
+        """
+        build_info = bm.get_singleton().get_build_info(build_id=build_id)
+
+        if build_info is None:
+            raise ValueError(f"No build found with ID {build_id}")
+
+        if build_info.progress.state != BuildState.RUNNING:
+            raise RuntimeError(
+                "This method should only be called for running builds."
+            )
+
+        # Builder ships the archive post completion
+        # This is irrespective of SUCCESS or FAILURE
+        if not os.path.exists(
+            bm.get_singleton().get_build_archive_path(build_id)
+        ):
+            return BuildState.RUNNING
+
+        log_file_path = bm.get_singleton().get_build_log_path(build_id)
+        try:
+            # Read the log content
+            with open(log_file_path, encoding='utf-8') as f:
+                build_log = f.read()
+        except FileNotFoundError:
+            self.logger.error(
+                f"Log file not found for RUNNING build with ID: {build_id}"
+            )
+            return BuildState.ERROR
+
+        # Build has finished, check if it succeeded or failed
+        success_message_pos = build_log.find(
+            f"'{build_info.vehicle.lower()}' finished successfully"
+        )
+        if success_message_pos == -1:
+            return BuildState.FAILURE
+        else:
+            return BuildState.SUCCESS
+
+    def __update_build_percent(self, build_id: str) -> None:
+        """
+        Update the progress percentage of a given build.
+        """
+        build_info = bm.get_singleton().get_build_info(build_id=build_id)
+
+        if build_info is None:
+            raise ValueError(f"No build found with ID {build_id}")
+
+        current_state = build_info.progress.state
+        current_percent = build_info.progress.percent
+        new_percent = current_percent
+        self.logger.debug(
+            f"Build id: {build_id}, "
+            f"Current state: {current_state}, "
+            f"Current percentage: {current_percent}, "
+        )
+        if current_state == BuildState.PENDING:
+            # Keep existing percentage
+            pass
+        elif current_state == BuildState.RUNNING:
+            new_percent = self.__calc_running_build_progress_percent(build_id)
+        elif current_state == BuildState.SUCCESS:
+            new_percent = 100
+        elif current_state == BuildState.FAILURE:
+            # Keep existing percentage
+            pass
+        elif current_state == BuildState.ERROR:
+            # Keep existing percentage
+            pass
+        else:
+            raise Exception("Unhandled BuildState.")
+
+        self.logger.debug(
+            f"Build id: {build_id}, "
+            f"New percentage: {new_percent}, "
+        )
+        if new_percent != current_percent:
+            bm.get_singleton().update_build_progress_percent(
+                build_id=build_id,
+                percent=new_percent
+            )
+
+    def __update_build_state(self, build_id: str) -> None:
+        """
+        Update the state of a given build.
+        """
+        build_info = bm.get_singleton().get_build_info(build_id=build_id)
+
+        if build_info is None:
+            raise ValueError(f"No build found with ID {build_id}")
+
+        current_state = build_info.progress.state
+        new_state = current_state
+        self.logger.debug(
+            f"Build id: {build_id}, "
+            f"Current state: {current_state.name}, "
+        )
+
+        log_file_path = bm.get_singleton().get_build_log_path(build_id)
+        if current_state == BuildState.PENDING:
+            # Builder creates log file when it starts
+            # running a build
+            if os.path.exists(log_file_path):
+                new_state = BuildState.RUNNING
+        elif current_state == BuildState.RUNNING:
+            new_state = self.__refresh_running_build_state(build_id)
+        elif current_state == BuildState.SUCCESS:
+            # SUCCESS is a conclusive state
+            pass
+        elif current_state == BuildState.FAILURE:
+            # FAILURE is a conclusive state
+            pass
+        elif current_state == BuildState.ERROR:
+            # ERROR is a conclusive state
+            pass
+        else:
+            raise Exception("Unhandled BuildState.")
+
+        self.logger.debug(
+            f"Build id: {build_id}, "
+            f"New state: {new_state.name}, "
+        )
+        if current_state != new_state:
+            bm.get_singleton().update_build_progress_state(
+                build_id=build_id,
+                new_state=new_state,
+            )
+
+    def __update_build_progress_all(self) -> None:
+        """
+        Update progress for all builds.
+
+        This method will iterate through all  builds, calculate their
+        progress, and update the build manager with the latest progress state
+        and percentage.
+        """
+        for build_id in bm.get_singleton().get_all_build_ids():
+            self.__update_build_state(build_id)
+            self.__update_build_percent(build_id)
+
+        # Generate status.json after updating build progress.
+        self.__generate_status_json()
+
+    def get_status_json_path(self) -> str:
+        """
+        Path to status.json file.
+        """
+        return os.path.join(
+            bm.get_singleton().get_outdir(),
+            'status.json'
+        )
+
+    def __ensure_status_json(self) -> None:
+        """
+        Ensures status.json exists and is a valid JSON file.
+        """
+        p = Path(self.get_status_json_path())
+
+        if not p.exists():
+            # Ensure parent directory exists
+            Path.mkdir(p.parent, parents=True, exist_ok=True)
+
+            # write empty json dict
+            with open(p, 'w') as f:
+                f.write('{}')
+
+    def __generate_status_json(self) -> None:
+        """
+        Rewrite status.json file.
+        """
+        all_build_ids_sorted = sorted(
+            bm.get_singleton().get_all_build_ids(),
+            key=lambda x: bm.get_singleton().get_build_info(x).time_created,
+            reverse=True
+        )
+
+        self.logger.debug(f"All build ids sorted: {all_build_ids_sorted}")
+        # To-do: fix status.json structure,
+        # write a list instead of a dict to the file
+        builds_dict = {}
+        for build_id in all_build_ids_sorted:
+            build_info = bm.get_singleton().get_build_info(build_id)
+            build_age_min = int(time.time() - build_info.time_created) // 60
+            bi_json = {
+                'vehicle': build_info.vehicle.capitalize(),
+                'board': build_info.board,
+                'git_hash_short': build_info.git_hash[:8],
+                'features': ', '.join(build_info.selected_features),
+                'status': build_info.progress.state.name,
+                'progress': build_info.progress.percent,
+                'age': "%u:%02u" % ((build_age_min // 60), build_age_min % 60)
+            }
+            self.logger.debug(f"Build info json: {bi_json}")
+            builds_dict[build_id] = bi_json
+        self.logger.debug(f"Builds dict: {builds_dict}")
+
+        with open(self.get_status_json_path(), 'w') as f:
+            f.write(json.dumps(builds_dict))
+
+    @staticmethod
+    def get_singleton() -> "BuildProgressUpdater":
+        """
+        Get the singleton instance of BuildProgressUpdater.
+
+        Returns:
+            BuildProgressUpdater: The singleton instance of this class.
+        """
+        return BuildProgressUpdater.__singleton