diff --git a/.gitignore b/.gitignore index c83c7c0d5e..d5802daa0c 100644 --- a/.gitignore +++ b/.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 diff --git a/packages/helpermodules/filesystem_migrations.py b/packages/helpermodules/filesystem_migrations.py new file mode 100644 index 0000000000..b359a33217 --- /dev/null +++ b/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_() 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() diff --git a/packages/helpermodules/filesystem_migrations_test.py b/packages/helpermodules/filesystem_migrations_test.py new file mode 100644 index 0000000000..9005449985 --- /dev/null +++ b/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() diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py index f19a3b1283..a360523825 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py +++ b/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() diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py index 81d03cb521..e7c7e0a3dd 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py @@ -1,43 +1,84 @@ +import fcntl from unittest.mock import Mock -from helpermodules.measurement_logging import generate_totals_subprocess +import pytest +from helpermodules.measurement_logging import generate_totals_subprocess -def test_generate_totals_calls_generator_when_lock_is_acquired(monkeypatch, tmp_path): - # setup - lock_file = tmp_path / "locks" / "generate_totals.lock" - monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - flock_mock = Mock() - generate_mock = Mock() +@pytest.mark.parametrize("fails", [False, True]) +def test_generate_totals_preserves_lock_file_and_releases_lock(monkeypatch, tmp_path, fails): + lock_path = tmp_path / "locks" / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_path) + locked_inode = [] - monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", flock_mock) - monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) + def generate(): + locked_inode.append(lock_path.stat().st_ino) + with lock_path.open("a") as contender: + with pytest.raises(BlockingIOError): + fcntl.flock(contender.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + if fails: + raise RuntimeError("generator failed") - # execution - generate_totals_subprocess.generate_totals() + generator = Mock(side_effect=generate) + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generator) - # evaluation - flock_mock.assert_called_once() - generate_mock.assert_called_once() - assert lock_file.parent.is_dir() + if fails: + with pytest.raises(RuntimeError, match="generator failed"): + generate_totals_subprocess.generate_totals() + else: + generate_totals_subprocess.generate_totals() + generator.assert_called_once_with() + assert lock_path.stat().st_ino == locked_inode[0] + with lock_path.open("a") as next_owner: + fcntl.flock(next_owner.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) -def test_generate_totals_skips_generator_when_lock_is_not_acquired(monkeypatch, tmp_path): - # setup - lock_file = tmp_path / "locks" / "generate_totals.lock" - monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - flock_mock = Mock(side_effect=RuntimeError("already locked")) - generate_mock = Mock() +def test_generate_totals_raises_without_generating_when_lock_is_contended(monkeypatch, tmp_path): + lock_path = tmp_path / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_path) + generator = Mock() + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generator) - monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", flock_mock) - monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) + with lock_path.open("a") as owner: + original_inode = lock_path.stat().st_ino + fcntl.flock(owner.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + with pytest.raises(BlockingIOError): + generate_totals_subprocess.generate_totals() + generator.assert_not_called() + assert lock_path.stat().st_ino == original_inode - # execution generate_totals_subprocess.generate_totals() + generator.assert_called_once_with() + assert lock_path.stat().st_ino == original_inode + + +@pytest.mark.parametrize("failure", [None, RuntimeError("broken log")]) +@pytest.mark.parametrize("period", ["daily", "monthly"]) +def test_generate_totals_continues_after_failed_logs(monkeypatch, mock_pub, period, failure): + days = [generate_totals_subprocess.Path(f"20220{month}01.json") for month in range(1, 4)] + monkeypatch.setattr(generate_totals_subprocess, "get_all_days_to_calc", lambda: days) + daily = Mock(return_value={}) + monthly = Mock(return_value={}) + failed = daily if period == "daily" else monthly + failed.side_effect = [failure, {}, {}] + log_error = Mock() + monkeypatch.setattr(generate_totals_subprocess.log, "exception", log_error) + monkeypatch.setattr(generate_totals_subprocess, "save_daily_source_totals", daily) + monkeypatch.setattr(generate_totals_subprocess, "save_monthly_source_totals", monthly) + + def publish(*args, **kwargs): + assert daily.call_count == 3 + assert monthly.call_count == 3 + return Mock() + + mock_pub.publisher.client.publish.side_effect = publish + + generate_totals_subprocess._generate_totals() - # evaluation - flock_mock.assert_called_once() - generate_mock.assert_not_called() - assert lock_file.parent.is_dir() + assert daily.call_args_list == [((day.stem,), {"saving": True}) for day in days] + assert monthly.call_args_list == [((f"20220{month}",), {"saving": True}) for month in range(1, 4)] + assert log_error.call_count == (1 if failure is not None else 0) + mock_pub.publisher.client.publish.assert_called_once_with( + "openWB/system/log_data_ready", "true", qos=1, retain=True) diff --git a/packages/helpermodules/setdata.py b/packages/helpermodules/setdata.py index 256e34d37c..aa975a8f51 100644 --- a/packages/helpermodules/setdata.py +++ b/packages/helpermodules/setdata.py @@ -12,6 +12,7 @@ from control import data from helpermodules import hardware_configuration, subdata from helpermodules.broker import BrokerClient +from helpermodules.filesystem_migrations import FilesystemMigrations from helpermodules.pub import Pub from helpermodules.utils.topic_parser import decode_payload, get_index, get_index_position from helpermodules.update_config import UpdateConfig @@ -1119,7 +1120,7 @@ def process_system_topic(self, msg: mqtt.MQTTMessage): elif "openWB/set/system/datastore_version" in msg.topic: self._validate_value(msg, int, [(0, UpdateConfig.DATASTORE_VERSION)], collection=list) elif "openWB/set/system/file_operation_version" in msg.topic: - self._validate_value(msg, int, [(0, UpdateConfig.FILE_OPERATION_VERSION)], collection=list) + self._validate_value(msg, int, [(0, FilesystemMigrations.FILE_OPERATION_VERSION)], collection=list) elif "openWB/set/system/GetRemoteSupport" in msg.topic: # Server-Topic enthält kein json-Payload. payload = msg.payload.decode("utf-8") diff --git a/packages/helpermodules/update_config.py b/packages/helpermodules/update_config.py index de5f64bc1a..c9aad61d10 100644 --- a/packages/helpermodules/update_config.py +++ b/packages/helpermodules/update_config.py @@ -65,8 +65,6 @@ class UpdateConfig: DATASTORE_VERSION = 156 - FILE_OPERATION_VERSION = 0 - valid_topic = [ "^openWB/bat/config/bat_control_activated$", "^openWB/bat/config/power_limit_mode$", @@ -834,11 +832,11 @@ def update(self): try: # erst breaking changes auflösen, sonst sind alte Topics schon gelöscht self.__solve_breaking_changes() - self.__solve_breaking_changes_filesystem() self.__remove_outdated_topics() self._remove_invalid_topics() self.__pub_missing_defaults() self.__update_version() + self.__start_filesystem_migrations() except Exception: log.exception("Fehler bei der Aktualisierung des Brokers.") pub_system_message({}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) @@ -935,66 +933,21 @@ def __solve_breaking_changes(self) -> None: pub_system_message( {}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) - def __solve_breaking_changes_filesystem(self) -> None: - """Führt dateisystembezogene Migrationen anhand der file_operation_version aus.""" - 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-Format (int): bereits ausgeführte Upgrades als Liste abbilden. - 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)): - try: - operation_required = version not in file_operation_version - if version == 0: - # Version 0 erneut ausführen, solange die Hintergrund-Generierung der Tages-/Monatssummen - # noch nicht erfolgreich abgeschlossen ist (Flag ist nicht True). - 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.debug(f"upgrading File Operation version '{version}'") - getattr(self, f"upgrade_file_operation_{version}")() - except AttributeError: - log.error(f"missing upgrade function! '{version}'") - except Exception: - log.exception("Fehler bei der Aktualisierung des Brokers.") - pub_system_message( - {}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) - - def upgrade_file_operation_0(self) -> None: + def __start_filesystem_migrations(self) -> None: """ - Generiere die Totals-Summen für Tage und Monate + Startet in einem eigen Prozess die Filesystem_Migration + -> alle folgenden upgrade_file_operation_X werden nacheinander und nicht parallel abgearbeitet """ - - self.__update_topic("openWB/system/log_data_ready", False) - try: - _generate_totals_subprocess = subprocess.Popen( - [ - sys.executable, - "-m", - "helpermodules.measurement_logging.generate_totals_subprocess" - ], - cwd="/var/www/html/openWB/packages", - start_new_session=True, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - close_fds=True - ) - - log.debug("generate_totals_subprocess gestartet, PID: %s", _generate_totals_subprocess.pid) - self._append_file_operation_version(0) - except Exception: - _generate_totals_subprocess = None - log.exception("Fehler beim Starten des generate_totals_subprocess.") + process = subprocess.Popen( + [sys.executable, "-m", "helpermodules.filesystem_migrations"], + cwd=self.base_path / "packages", + start_new_session=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=True, + ) + log.debug("Dateisystem-Migrationen gestartet, PID: %s", process.pid) def _loop_all_received_topics(self, callback) -> None: modified_topics = {} @@ -1014,12 +967,6 @@ def _append_datastore_version(self, version: int) -> None: datastore_versions.append(version) self.__update_topic("openWB/system/datastore_version", datastore_versions) - def _append_file_operation_version(self, version: int) -> None: - file_operation_versions = 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 upgrade_datastore_0(self) -> None: def upgrade(topic: str, payload) -> Optional[dict]: modified_topics = {} diff --git a/packages/helpermodules/update_config_test.py b/packages/helpermodules/update_config_test.py index aee711d241..d2117ec280 100644 --- a/packages/helpermodules/update_config_test.py +++ b/packages/helpermodules/update_config_test.py @@ -619,23 +619,6 @@ def test_upgrade_datastore_153_migrates_legacy_modes( assert uc.all_received_topics["openWB/system/datastore_version"] == [152, 153] -@pytest.mark.parametrize("file_operation_version, finished, expected_calls", [ - ([], False, 1), # erster Start - ([0], True, 0), # bereits fertig -> kein Neustart - ([0], False, 1), # angefangen, aber nicht fertig -> Neustart -]) -def test_file_operation_0_start_behavior(file_operation_version, finished, expected_calls): - update_config = UpdateConfig() - update_config.all_received_topics = { - "openWB/system/file_operation_version": file_operation_version, - "openWB/system/log_data_ready": finished - } - - with patch.object(update_config, "upgrade_file_operation_0") as upgrade_mock: - update_config._UpdateConfig__solve_breaking_changes_filesystem() - assert upgrade_mock.call_count == expected_calls - - @pytest.mark.parametrize("provider_type, configuration, expected_configuration, updated", [ ("energycharts", {"surcharge": 0.0001}, {"surcharge": 0.0001, "price_cap": None}, True), ("energycharts", {"price_cap": 0.0002}, {"price_cap": 0.0002}, False), @@ -666,3 +649,21 @@ def test_upgrade_datastore_156_adds_energycharts_price_cap( updated_topics = [call.args[0] for call in mock_pub.pub.call_args_list] assert (topic in updated_topics) is updated assert uc.all_received_topics["openWB/system/datastore_version"] == [155, 156] + + +def test_start_filesystem_migrations_launches_one_worker(monkeypatch): + updater = UpdateConfig() + popen = Mock() + monkeypatch.setattr(update_config.subprocess, "Popen", popen) + + updater._UpdateConfig__start_filesystem_migrations() + + popen.assert_called_once_with( + [update_config.sys.executable, "-m", "helpermodules.filesystem_migrations"], + cwd=updater.base_path / "packages", + start_new_session=True, + stdin=update_config.subprocess.DEVNULL, + stdout=update_config.subprocess.DEVNULL, + stderr=update_config.subprocess.DEVNULL, + close_fds=True, + )