refactor(puller): Create database driver from factory in poller thread

Use a readable database factory in the puller actor and create the
concrete driver from the database poller thread, so database resources
are initialized inside the actor process.

Author: gfieni

src/powerapi/puller/database_poller.py

@@ -36,10 +36,10 @@
 from time import sleep
 from typing import TYPE_CHECKING
 
-from powerapi.database import ConnectionFailed, ReadFailed
+from powerapi.database.exceptions import ConnectionFailed, ReadFailed
 
 if TYPE_CHECKING:
-    from powerapi.database import ReadableDatabase
+    from powerapi.database.driver import ReadableDatabaseFactory
     from powerapi.filter import ReportFilter
 
 
@@ -49,16 +49,16 @@ class DatabasePollerThread(Thread):
     Fetches reports from a database and forward it to theirs corresponding dispatcher.
     """
 
-    def __init__(self, database: ReadableDatabase, report_filter: ReportFilter, stream_mode: bool, poll_interval: float = 1.0):
+    def __init__(self, database_factory: ReadableDatabaseFactory, report_filter: ReportFilter, stream_mode: bool, poll_interval: float = 1.0):
         """
-        :param database: Database to retrieve the reports from
+        :param database_factory: Factory used to create the database driver
         :param report_filter: Report filter used to dispatch the received reports
         :param stream_mode: If true, poll continuously from the database; otherwise, stop the poller thread on empty result
         :param poll_interval: Interval in seconds between database polls
         """
         super().__init__(name='database-poller-thread', daemon=True)
 
-        self.database = database
+        self.database_factory = database_factory
         self.report_filter = report_filter
         self.stream_mode = stream_mode
         self.poll_interval = poll_interval
@@ -87,7 +87,11 @@ def run(self) -> None:
         Entrypoint of the database poller thread.
         """
         try:
-            self.database.connect()
+            database = self.database_factory.create()
+            database.connect()
+        except ValueError as exn:
+            logging.error('Failed to create the database driver: %s', exn)
+            return
         except ConnectionFailed as exn:
             logging.error('Failed to connect the database driver: %s', exn.msg)
             return
@@ -100,7 +104,7 @@ def run(self) -> None:
 
         while not self._stop_event.is_set():
             try:
-                for report in self.database.read(self.stream_mode):
+                for report in database.read(self.stream_mode):
                     for dispatcher in self.report_filter.route(report):
                         dispatcher.send_data(report)
 
@@ -115,7 +119,7 @@ def run(self) -> None:
         for dispatcher in self.report_filter.dispatchers():
             dispatcher.disconnect()
 
-        self.database.disconnect()
+        database.disconnect()
         logging.info('Database poller thread stopped')
 
     def wait_ready(self, timeout: float | None = None) -> bool:

src/powerapi/puller/puller_actor.py

@@ -31,7 +31,7 @@
 
 from powerapi.actor import Actor, State
 from powerapi.actor.message import StartMessage, PoisonPillMessage
-from powerapi.database import ReadableDatabase
+from powerapi.database.driver import ReadableDatabaseFactory
 from powerapi.filter import ReportFilter
 from powerapi.puller.database_poller import DatabasePollerThread
 from powerapi.puller.handlers import PullerStartMessageHandler, PullerPoisonPillMessageHandler
@@ -42,20 +42,20 @@ class PullerState(State):
     Puller Actor State class.
     """
 
-    def __init__(self, actor: Actor, database: ReadableDatabase, report_filter: ReportFilter, stream_mode: bool):
+    def __init__(self, actor: Actor, database_factory: ReadableDatabaseFactory, report_filter: ReportFilter, stream_mode: bool):
         """
         :param actor: Puller actor instance
-        :param database: Database driver
+        :param database_factory: Factory used to create the database driver
         :param report_filter: Filter to use when dispatching reports
         :param stream_mode: If true, poll continuously from the database; otherwise, stop the poller thread on empty result
         """
         super().__init__(actor)
 
-        self.database = database
+        self.database_factory = database_factory
         self.report_filter = report_filter
         self.stream_mode = stream_mode
 
-        self.db_poller_thread = DatabasePollerThread(database, report_filter, stream_mode)
+        self.db_poller_thread = DatabasePollerThread(database_factory, report_filter, stream_mode)
 
 
 class PullerActor(Actor):
@@ -64,24 +64,24 @@ class PullerActor(Actor):
     This actor allows to retrieve reports from a database and send them to theirs corresponding dispatcher.
     """
 
-    def __init__(self, name: str, database: ReadableDatabase, report_filter: ReportFilter, stream_mode: bool = False, level_logger: int = logging.WARNING):
+    def __init__(self, name: str, database_factory: ReadableDatabaseFactory, report_filter: ReportFilter, stream_mode: bool = False, level_logger: int = logging.WARNING):
         """
         :param name: Name of the puller actor
-        :param database: Database driver to use to persist reports
+        :param database_factory: Factory used to create the database driver
         :param report_filter: Filter to use when dispatching reports
         :param level_logger: Define the level of the logger for the actor
         """
         super().__init__(name, level_logger, 1000)
 
-        self.database = database
+        self.database_factory = database_factory
         self.report_filter = report_filter
         self.stream_mode = stream_mode
 
     def setup(self) -> None:
         """
         Set up the Puller actor message handlers.
         """
-        self.state = PullerState(self, self.database, self.report_filter, self.stream_mode)
+        self.state = PullerState(self, self.database_factory, self.report_filter, self.stream_mode)
 
         self.add_handler(StartMessage, PullerStartMessageHandler(self.state))
         self.add_handler(PoisonPillMessage, PullerPoisonPillMessageHandler(self.state))

tests/unit/puller/conftest.py

@@ -36,6 +36,7 @@
 from powerapi.filter import BroadcastReportFilter
 from powerapi.puller.handlers import PullerStartMessageHandler, PullerPoisonPillMessageHandler
 from powerapi.puller.puller_actor import PullerState
+from tests.utils.db import PrebuiltDatabaseFactory
 
 if TYPE_CHECKING:
     from powerapi.filter import ReportFilter
@@ -88,8 +89,9 @@ def puller_start_handler(make_fake_failing_database, fake_database_poller):
     def _create_handler(report_filter: ReportFilter) -> PullerStartMessageHandler:
         actor = Mock()
         database = make_fake_failing_database()
+        database_factory = PrebuiltDatabaseFactory(database)
 
-        state = PullerState(actor, database, report_filter, stream_mode=False)
+        state = PullerState(actor, database_factory, report_filter, stream_mode=False)
         state.db_poller_thread = fake_database_poller
 
         handler = PullerStartMessageHandler(state)
@@ -108,8 +110,9 @@ def _create_handler() -> PullerPoisonPillMessageHandler:
         actor.socket_interface.receive.return_value = None  # Prevents an infinite loop when triggering a graceful shutdown.
 
         database = make_fake_failing_database()
+        database_factory = PrebuiltDatabaseFactory(database)
 
-        state = PullerState(actor, database, empty_report_filter, stream_mode=False)
+        state = PullerState(actor, database_factory, empty_report_filter, stream_mode=False)
         state.db_poller_thread = fake_database_poller
 
         handler = PullerPoisonPillMessageHandler(state)