Update Filesystem_Migration

.gitignore

@@ -19,6 +19,8 @@ data/daily_log/*
 data/daily_totals/*
 data/monthly_log/*
 data/monthly_totals/*
+data/filesystem_migrations.lock
+data/generate_totals.lock
 data/backup/*.tar.gz
 data/backup/*.openwb-backup*
 data/restore/*.tar.gz

packages/helpermodules/filesystem_migrations.py

@@ -0,0 +1,124 @@
+# UpdateConfig startet dieses Modul als eigenen Hintergrundprozess. Eine globale
+# Dateisperre verhindert parallele Worker. Nach dem Einlesen des MQTT-Zustands
+# laufen fehlende upgrade_file_operation_<version>() synchron in Versionsreihenfolge.
+# Erst nach normaler Rückkehr wird die Version gespeichert; bei einer weitergegebenen
+# Exception endet der Lauf. Beim nächsten Worker-Start werden abgeschlossene Versionen
+# übersprungen und die fehlgeschlagene Migration erneut versucht (kein Rollback).
+# Migrationen müssen deshalb nach Teilabbrüchen wiederholbar sein. Neue Migrationen
+# hier ergänzen und FILE_OPERATION_VERSION anheben.
+# log_data_ready betrifft nur Migration 0, nicht den Abschluss aller Migrationen.
+
+import copy
+import fcntl
+import json
+import logging
+from pathlib import Path
+
+from paho.mqtt.client import Client as MqttClient, MQTTMessage
+
+from helpermodules import logger
+from helpermodules.broker import BrokerClient
+from helpermodules.measurement_logging.generate_totals_subprocess import generate_totals
+from helpermodules.messaging import MessageType, pub_system_message
+from helpermodules.pub import Pub
+from helpermodules.utils.topic_parser import decode_payload
+
+
+log = logging.getLogger(__name__)
+LOCK_FILE = Path(__file__).resolve().parents[2] / "data" / "filesystem_migrations.lock"
+
+
+class FilesystemMigrations:
+    FILE_OPERATION_VERSION = 0
+
+    def __init__(self) -> None:
+        self.all_received_topics = {}
+
+    def on_connect(self, client: MqttClient, userdata, flags: dict, rc: int):
+        client.subscribe("openWB/#", 2)
+
+    def on_message(self, client: MqttClient, userdata, msg: MQTTMessage):
+        self.all_received_topics.update({msg.topic: msg.payload})
+
+    def _update_topic(self, topic: str, payload) -> None:
+        publish_info = Pub().publisher.client.publish(topic, json.dumps(payload), qos=1, retain=True)
+        publish_info.wait_for_publish(timeout=10)
+        if not publish_info.is_published():
+            raise TimeoutError(f"Topic '{topic}' konnte nicht gespeichert werden.")
+        self.all_received_topics[topic] = copy.deepcopy(payload)
+
+    def update_filesystem(self) -> None:
+        """Liest den aktuellen Broker-Zustand im Migrationsprozess ein."""
+        broker = BrokerClient("filesystem-migrations", self.on_connect, self.on_message)
+        try:
+            broker.start_finite_loop()
+        finally:
+            broker.disconnect()
+        self.__solve_breaking_changes_filesystem()
+
+    def __solve_breaking_changes_filesystem(self) -> None:
+        """Führt fehlende Migrationen synchron aus und markiert nur erfolgreiche Abschlüsse."""
+        file_operation_version = decode_payload(self.all_received_topics.get("openWB/system/file_operation_version"))
+        if file_operation_version is None:
+            # Neues Topic (z.B. bei Upgrade von älteren Versionen): alle File-Operation-Upgrades ausführen.
+            file_operation_version = []
+            self._update_topic("openWB/system/file_operation_version", file_operation_version)
+        elif isinstance(file_operation_version, int):
+            # Legacy-Zähler: Die Versionen vor diesem Wert sind bereits abgeschlossen.
+            file_operation_version = list(range(file_operation_version))
+            self._update_topic("openWB/system/file_operation_version", file_operation_version)
+
+        log.debug(f"current file operation version: {file_operation_version}")
+        log.debug(f"target file operation version: {self.FILE_OPERATION_VERSION}")
+        for version in list(range(self.FILE_OPERATION_VERSION+1)):
+            operation_required = version not in file_operation_version
+            if version == 0:
+                # Alte Installationen haben Version 0 schon beim Prozessstart eingetragen.
+                log_totals_generation_finished = decode_payload(
+                    self.all_received_topics.get("openWB/system/log_data_ready"))
+                operation_required = operation_required or log_totals_generation_finished is not True
+
+            if operation_required:
+                log.info("Dateisystem-Migration %s wird ausgeführt.", version)
+                getattr(self, f"upgrade_file_operation_{version}")()
+                self._append_file_operation_version(version)
+
+    def upgrade_file_operation_0(self) -> None:
+        """Generiere die Totals-Summen für Tage und Monate."""
+
+        self._update_topic("openWB/system/log_data_ready", False)
+        generate_totals()
+        self.all_received_topics["openWB/system/log_data_ready"] = True
+
+    def _append_file_operation_version(self, version: int) -> None:
+        file_operation_versions = list(decode_payload(
+            self.all_received_topics.get("openWB/system/file_operation_version")))
+        if version not in file_operation_versions:
+            file_operation_versions.append(version)
+            self._update_topic("openWB/system/file_operation_version", file_operation_versions)
+
+
+def run_migrations() -> None:
+    LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
+    # Lock-Datei beibehalten, damit alle Prozesse denselben Inode sperren.
+    # Löschen und Neuanlegen könnte parallele Läufe ermöglichen.
+    # Die Sperre wird beim Verlassen des with-Blocks automatisch freigegeben.
+    with LOCK_FILE.open("a") as lock_file:
+        try:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            log.info("Dateisystem-Migrationen laufen bereits. Kein zweiter Lauf gestartet.")
+            return
+
+        try:
+            FilesystemMigrations().update_filesystem()
+        except Exception:
+            log.exception("Dateisystem-Migrationen abgebrochen.")
+            pub_system_message({}, "Fehler bei der Aktualisierung der Dateisystem-Daten.", MessageType.ERROR)
+            raise
+
+
+if __name__ == "__main__":
+    logger.setup_generate_totals_logging(include_process_log=True)
+    logging.basicConfig(level=logging.DEBUG, handlers=logging.getLogger("generate_totals").handlers)
+    run_migrations()

packages/helpermodules/filesystem_migrations_test.py

@@ -0,0 +1,196 @@
+import json
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import pytest
+from paho.mqtt.client import MQTTMessage
+
+from helpermodules import filesystem_migrations
+from helpermodules.filesystem_migrations import FilesystemMigrations
+
+
+@pytest.mark.parametrize("file_operation_version, finished, expected_calls", [
+    ([], False, 1),
+    ([0], True, 0),
+    ([0], False, 1),
+    (None, None, 1),
+])
+def test_file_operation_0_start_behavior(monkeypatch, file_operation_version, finished, expected_calls):
+    monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 0)
+    updater = FilesystemMigrations()
+    updater.all_received_topics = {
+        "openWB/system/file_operation_version": file_operation_version,
+        "openWB/system/log_data_ready": finished,
+    }
+
+    with patch.object(updater, "upgrade_file_operation_0") as upgrade_mock:
+        updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+    assert upgrade_mock.call_count == expected_calls
+
+
+@pytest.mark.parametrize("progress", [0, 1, 2])
+def test_legacy_integer_progress_is_persisted_as_list_and_pending_migrations_run(monkeypatch, mock_pub, progress):
+    updater = FilesystemMigrations()
+    updater.all_received_topics = {
+        "openWB/system/file_operation_version": str(progress).encode(),
+        "openWB/system/log_data_ready": b"true",
+    }
+    monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 2)
+    calls = []
+    for version in range(3):
+        monkeypatch.setattr(
+            updater, f"upgrade_file_operation_{version}",
+            lambda version=version: calls.append(version), raising=False,
+        )
+
+    updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+
+    assert calls == list(range(progress, 3))
+    assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1, 2]
+    first_publish = mock_pub.publisher.client.publish.call_args_list[0]
+    assert first_publish.args == ("openWB/system/file_operation_version", json.dumps(list(range(progress))))
+    assert first_publish.kwargs == {"qos": 1, "retain": True}
+
+
+def test_file_operations_run_sequentially_and_resume_after_failure(monkeypatch):
+    fails = True
+    updater = FilesystemMigrations()
+    updater.all_received_topics = {
+        "openWB/system/file_operation_version": [0],
+        "openWB/system/log_data_ready": True,
+    }
+    monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 3)
+    calls = []
+
+    def migration_1():
+        assert updater.all_received_topics["openWB/system/file_operation_version"] == [0]
+        calls.append(1)
+
+    def migration_2():
+        assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1]
+        calls.append(2)
+        if fails:
+            raise RuntimeError("migration failed")
+
+    def migration_3():
+        assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1, 2]
+        calls.append(3)
+
+    monkeypatch.setattr(updater, "upgrade_file_operation_1", migration_1, raising=False)
+    monkeypatch.setattr(updater, "upgrade_file_operation_2", migration_2, raising=False)
+    monkeypatch.setattr(updater, "upgrade_file_operation_3", migration_3, raising=False)
+
+    with pytest.raises(RuntimeError, match="migration failed"):
+        updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+    assert calls == [1, 2]
+    assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1]
+    fails = False
+    updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+    assert calls == [1, 2, 2, 3]
+
+    assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1, 2, 3]
+
+
+def test_file_operation_stops_if_version_cannot_be_published(monkeypatch, mock_pub):
+    monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1)
+    updater = FilesystemMigrations()
+    updater.all_received_topics = {
+        "openWB/system/file_operation_version": [],
+        "openWB/system/log_data_ready": True,
+    }
+    monkeypatch.setattr(updater, "upgrade_file_operation_0", Mock())
+    next_migration = Mock()
+    monkeypatch.setattr(updater, "upgrade_file_operation_1", next_migration, raising=False)
+    mock_pub.publisher.client.publish.return_value.is_published.return_value = False
+
+    with pytest.raises(TimeoutError, match="konnte nicht gespeichert"):
+        updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+
+    assert updater.all_received_topics["openWB/system/file_operation_version"] == []
+    next_migration.assert_not_called()
+
+
+def test_next_migration_runs_after_all_totals_attempts_despite_failed_day(monkeypatch, mock_pub):
+    monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1)
+    from helpermodules.measurement_logging import generate_totals_subprocess
+
+    updater = FilesystemMigrations()
+    updater.all_received_topics = {
+        "openWB/system/file_operation_version": [],
+        "openWB/system/log_data_ready": False,
+    }
+    days = [Path("20220101.json"), Path("20220102.json")]
+    monkeypatch.setattr(generate_totals_subprocess, "get_all_days_to_calc", lambda: days)
+    daily = Mock(side_effect=[RuntimeError("broken log"), {}])
+    monthly = Mock(return_value={})
+    monkeypatch.setattr(generate_totals_subprocess, "save_daily_source_totals", daily)
+    monkeypatch.setattr(generate_totals_subprocess, "save_monthly_source_totals", monthly)
+    monkeypatch.setattr(filesystem_migrations, "generate_totals", generate_totals_subprocess._generate_totals)
+
+    def next_migration():
+        assert daily.call_count == 2
+        monthly.assert_called_once_with("202201", saving=True)
+        assert updater.all_received_topics["openWB/system/log_data_ready"] is True
+        assert updater.all_received_topics["openWB/system/file_operation_version"] == [0]
+        assert any(call.args == ("openWB/system/log_data_ready", "true")
+                   for call in mock_pub.publisher.client.publish.call_args_list)
+
+    next_operation = Mock(side_effect=next_migration)
+    monkeypatch.setattr(updater, "upgrade_file_operation_1", next_operation, raising=False)
+
+    updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+
+    next_operation.assert_called_once_with()
+    assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1]
+
+
+@pytest.mark.parametrize("completed, ready", [([], True), ([0], False)])
+def test_file_operation_0_failure_leaves_logs_unready_and_blocks_next(monkeypatch, mock_pub, completed, ready):
+    monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1)
+    updater = FilesystemMigrations()
+    updater.all_received_topics = {
+        "openWB/system/file_operation_version": completed,
+        "openWB/system/log_data_ready": ready,
+    }
+    monkeypatch.setattr(
+        filesystem_migrations, "generate_totals", Mock(side_effect=RuntimeError("generator failed")))
+    next_migration = Mock()
+    monkeypatch.setattr(updater, "upgrade_file_operation_1", next_migration, raising=False)
+
+    with pytest.raises(RuntimeError, match="generator failed"):
+        updater._FilesystemMigrations__solve_breaking_changes_filesystem()
+
+    assert updater.all_received_topics["openWB/system/log_data_ready"] is False
+    assert updater.all_received_topics["openWB/system/file_operation_version"] == completed
+    mock_pub.publisher.client.publish.assert_called_once_with(
+        "openWB/system/log_data_ready", "false", qos=1, retain=True)
+    next_migration.assert_not_called()
+
+
+def test_filesystem_worker_reads_current_topics_before_migrating(monkeypatch):
+    updater = FilesystemMigrations()
+    broker = Mock()
+    broker_client = Mock(return_value=broker)
+    monkeypatch.setattr(filesystem_migrations, "BrokerClient", broker_client)
+    topics = {
+        "openWB/system/file_operation_version": b"[0]",
+        "openWB/system/log_data_ready": b"true",
+    }
+
+    def read_topics():
+        on_message = broker_client.call_args.args[2]
+        for topic, payload in topics.items():
+            message = MQTTMessage(topic=topic.encode())
+            message.payload = payload
+            on_message(broker.client, None, message)
+        assert updater.all_received_topics == topics
+
+    broker.start_finite_loop.side_effect = read_topics
+    migration = Mock()
+    monkeypatch.setattr(updater, "upgrade_file_operation_0", migration)
+
+    updater.update_filesystem()
+
+    migration.assert_not_called()
+    broker.start_finite_loop.assert_called_once_with()
+    broker.disconnect.assert_called_once_with()

packages/helpermodules/measurement_logging/generate_totals_subprocess.py

@@ -13,7 +13,6 @@
 
 
 BASE_PATH = Path(__file__).resolve().parents[3]
-logger.setup_generate_totals_logging(include_process_log=True)
 log = logging.getLogger("generate_totals")
 
 LOCK_FILE = BASE_PATH / "data" / "generate_totals.lock"
@@ -73,33 +72,36 @@ def _generate_totals():
     log.info(f"Totals-Migration abgeschlossen. Fehlerhafte Logs: {errors}.")
     publisher = pub.Pub()
     publish_info = publisher.publisher.client.publish(
-        "openWB/set/system/log_data_ready", json.dumps(True), qos=1, retain=True)
+        "openWB/system/log_data_ready", json.dumps(True), qos=1, retain=True)
     publish_info.wait_for_publish()
 
 
 def generate_totals():
     LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
 
-    with LOCK_FILE.open("w") as lock_file:
+    # Lock-Datei beibehalten, damit alle Prozesse denselben Inode sperren.
+    # Löschen und Neuanlegen könnte parallele Läufe ermöglichen.
+    # Die Sperre wird beim Verlassen des with-Blocks automatisch freigegeben.
+    with LOCK_FILE.open("a") as lock_file:
         try:
             fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except Exception:
+        except BlockingIOError:
             log.exception("generate_totals_subprocess läuft bereits. "
                           "Der zweite Prozess wird beendet.")
-            return
+            raise
         try:
             current_lock_stat = LOCK_FILE.stat()
-        except FileNotFoundError:
-            return
+        except FileNotFoundError as exc:
+            raise RuntimeError(
+                "Lock-Datei wurde zwischenzeitlich gelöscht. Totals-Berechnung wird nicht gestartet."
+            ) from exc
         # Ein anderer Prozess kann die bereits geöffnete Lock-Datei inzwischen gelöscht und neu angelegt haben.
         # Unsere Sperre gilt dann nur für die alte Datei. Abbrechen, damit nicht zwei Prozesse gleichzeitig rechnen.
         if not os.path.samestat(current_lock_stat, os.fstat(lock_file.fileno())):
-            return
-        try:
-            _generate_totals()
-        finally:
-            LOCK_FILE.unlink(missing_ok=True)
+            raise RuntimeError("Lock-Datei der Totals-Migration wurde zwischenzeitlich ausgetauscht.")
+        _generate_totals()
 
 
 if __name__ == "__main__":
+    logger.setup_generate_totals_logging(include_process_log=True)
     generate_totals()