diff --git a/CHANGES/0000.bugfix b/CHANGES/0000.bugfix new file mode 100644 index 00000000000..9d7bd7acb7c --- /dev/null +++ b/CHANGES/0000.bugfix @@ -0,0 +1,3 @@ +API and content workers now tolerate a read-only database by skipping heartbeat writes +instead of crash-looping. Workers continue serving read traffic and resume normal +heartbeats once the database becomes writable. diff --git a/pulpcore/app/entrypoint.py b/pulpcore/app/entrypoint.py index ab76e7e0fec..5f2605db15b 100644 --- a/pulpcore/app/entrypoint.py +++ b/pulpcore/app/entrypoint.py @@ -18,6 +18,7 @@ PulpcoreGunicornApplication, handle_control_interface_feature, ) +from pulpcore.app.util import _is_read_only_db_error logger = getLogger(__name__) @@ -52,6 +53,27 @@ def notify(self): os._exit(Arbiter.WORKER_BOOT_ERROR) def heartbeat(self): + if self.app_status is None: + from pulpcore.app.models import AppStatus + + try: + self.app_status = AppStatus.objects.create( + name=self.name, app_type="api", versions=self.versions + ) + except IntegrityError: + logger.error( + f"An API app with name {self.name} already exists in the database." + ) + raise + except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + connection.close_if_unusable_or_obsolete() + logger.debug( + f"Api App '{self.name}' heartbeat skipped (read-only transaction)." + ) + return + raise + try: self.app_status.save_heartbeat() logger.debug(self.beat_msg) @@ -61,6 +83,11 @@ def heartbeat(self): self.app_status.save_heartbeat() logger.debug(self.beat_msg) except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + logger.warning( + f"Api App '{self.name}' heartbeat skipped (read-only transaction)." + ) + return logger.error(f"{self.fail_beat_msg} Exception: {str(e)}") raise @@ -101,6 +128,15 @@ def init_process(self): except IntegrityError: logger.error(f"An API app with name {self.name} already exists in the database.") exit(Arbiter.WORKER_BOOT_ERROR) + except DatabaseError as e: + if _is_read_only_db_error(e): + logger.warning( + f"Api App '{self.name}' could not register in the database " + "(read-only transaction). Heartbeat will retry." + ) + self.app_status = None + else: + raise self.heartbeat_interval = settings.API_APP_TTL // 3 @@ -115,9 +151,13 @@ def run(self): try: super().run() finally: - # cleanup if self.app_status: - self.app_status.delete() + try: + self.app_status.delete() + except (InterfaceError, DatabaseError): + logger.info( + f"Api App '{self.name}' could not clean up its status record." + ) class PulpcoreApiApplication(PulpcoreGunicornApplication): diff --git a/pulpcore/app/util.py b/pulpcore/app/util.py index 76f2a5b47fb..66e597d6076 100644 --- a/pulpcore/app/util.py +++ b/pulpcore/app/util.py @@ -677,6 +677,17 @@ def get_worker_name(): ) +def _is_read_only_db_error(exc): + """Check if a DatabaseError was caused by a read-only database (SQLSTATE 25006).""" + cause = exc.__cause__ + while cause is not None: + pgcode = getattr(cause, "pgcode", None) or getattr(cause, "sqlstate", None) + if pgcode == "25006": + return True + cause = getattr(cause, "__cause__", None) + return False + + def normalize_http_status(status): """Convert the HTTP status code to 2xx, 3xx, etc., normalizing the last two digits.""" if 100 <= status < 200: diff --git a/pulpcore/content/__init__.py b/pulpcore/content/__init__.py index decf0957b0d..0464bc356d7 100644 --- a/pulpcore/content/__init__.py +++ b/pulpcore/content/__init__.py @@ -22,7 +22,7 @@ from pulpcore.app.apps import pulp_plugin_configs # noqa: E402 from pulpcore.app.models import AppStatus # noqa: E402 -from pulpcore.app.util import get_worker_name # noqa: E402 +from pulpcore.app.util import _is_read_only_db_error, get_worker_name # noqa: E402 from .authentication import authenticate, guid # noqa: E402 from .handler import Handler # noqa: E402 @@ -57,6 +57,7 @@ async def _heartbeat(): ) versions = {app.label: app.version for app in pulp_plugin_configs()} + app_status = None try: app_status = await AppStatus.objects.acreate( name=name, app_type="content", versions=versions @@ -64,9 +65,38 @@ async def _heartbeat(): except IntegrityError: log.error(f"A content app with name {name} already exists in the database.") exit(Arbiter.WORKER_BOOT_ERROR) + except DatabaseError as e: + if _is_read_only_db_error(e): + log.warning( + f"Content App '{name}' could not register in the database " + "(read-only transaction). Heartbeat will retry." + ) + else: + raise + try: while True: await asyncio.sleep(heartbeat_interval) + + if app_status is None: + try: + app_status = await AppStatus.objects.acreate( + name=name, app_type="content", versions=versions + ) + except IntegrityError: + log.error( + f"A content app with name {name} already exists in the database." + ) + raise + except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + await sync_to_async(Handler._reset_db_connection)() + log.debug( + f"Content App '{name}' heartbeat skipped (read-only transaction)." + ) + continue + raise + try: await app_status.asave_heartbeat() log.debug(msg) @@ -76,11 +106,20 @@ async def _heartbeat(): await app_status.asave_heartbeat() log.debug(msg) except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + log.warning( + f"Content App '{name}' heartbeat skipped " + "(read-only transaction)." + ) + continue log.error(f"{fail_msg} Exception: {str(e)}") exit(Arbiter.WORKER_BOOT_ERROR) finally: if app_status: - await app_status.adelete() + try: + await app_status.adelete() + except (InterfaceError, DatabaseError): + log.info(f"Content App '{name}' could not clean up its status record.") async def _heartbeat_ctx(app): diff --git a/pulpcore/tests/unit/content/test_heartbeat.py b/pulpcore/tests/unit/content/test_heartbeat.py index 5fff2076431..9de5e48b9a3 100644 --- a/pulpcore/tests/unit/content/test_heartbeat.py +++ b/pulpcore/tests/unit/content/test_heartbeat.py @@ -1,7 +1,7 @@ -from unittest.mock import AsyncMock, Mock, call +from unittest.mock import AsyncMock, Mock, call, patch import pytest -from django.db.utils import InterfaceError, OperationalError +from django.db.utils import InternalError, InterfaceError, OperationalError from pulpcore.app.models.status import AppStatus, AppStatusManager from pulpcore.content import _heartbeat @@ -12,6 +12,14 @@ class MockException(Exception): pass +def _make_read_only_error(): + cause = Exception("cannot execute UPDATE in a read-only transaction") + cause.pgcode = "25006" + exc = InternalError("read-only") + exc.__cause__ = cause + return exc + + @pytest.mark.parametrize("error_class", [InterfaceError, OperationalError]) @pytest.mark.asyncio async def test_db_connection_interface_error(monkeypatch, settings, error_class): @@ -35,3 +43,68 @@ async def test_db_connection_interface_error(monkeypatch, settings, error_class) mock_app_status.asave_heartbeat.assert_called() mock_reset_db.assert_has_calls([call()]) + + +@pytest.mark.asyncio +async def test_read_only_heartbeat_skips_without_exit(monkeypatch, settings): + """ + When the DB is read-only, heartbeat should skip without exiting. + """ + read_only_exc = _make_read_only_error() + + mock_app_status = AsyncMock() + mock_app_status.asave_heartbeat.side_effect = [read_only_exc, read_only_exc] + mock_acreate = AsyncMock() + mock_acreate.return_value = mock_app_status + monkeypatch.setattr(AppStatusManager, "acreate", mock_acreate) + monkeypatch.setattr(AppStatus, "objects", AppStatusManager()) + mock_reset_db = Mock() + monkeypatch.setattr(Handler, "_reset_db_connection", mock_reset_db) + settings.CONTENT_APP_TTL = 1 + + iteration_count = 0 + + original_sleep = __import__("asyncio").sleep + + async def counting_sleep(seconds): + nonlocal iteration_count + iteration_count += 1 + if iteration_count >= 2: + raise KeyboardInterrupt("stop test loop") + await original_sleep(0) + + with patch("asyncio.sleep", side_effect=counting_sleep): + with pytest.raises(KeyboardInterrupt): + await _heartbeat() + + mock_app_status.asave_heartbeat.assert_called() + mock_reset_db.assert_called() + + +@pytest.mark.asyncio +async def test_read_only_acreate_skips_without_exit(monkeypatch, settings): + """ + When acreate fails with read-only, _heartbeat should not exit. + """ + read_only_exc = _make_read_only_error() + + mock_acreate = AsyncMock(side_effect=read_only_exc) + monkeypatch.setattr(AppStatusManager, "acreate", mock_acreate) + monkeypatch.setattr(AppStatus, "objects", AppStatusManager()) + mock_reset_db = Mock() + monkeypatch.setattr(Handler, "_reset_db_connection", mock_reset_db) + settings.CONTENT_APP_TTL = 1 + + iteration_count = 0 + + async def counting_sleep(seconds): + nonlocal iteration_count + iteration_count += 1 + if iteration_count >= 2: + raise KeyboardInterrupt("stop test loop") + + with patch("asyncio.sleep", side_effect=counting_sleep): + with pytest.raises(KeyboardInterrupt): + await _heartbeat() + + assert mock_acreate.call_count >= 2 diff --git a/pulpcore/tests/unit/test_entrypoint.py b/pulpcore/tests/unit/test_entrypoint.py new file mode 100644 index 00000000000..fc0a5b53d54 --- /dev/null +++ b/pulpcore/tests/unit/test_entrypoint.py @@ -0,0 +1,88 @@ +from unittest.mock import MagicMock, patch + +import pytest +from django.db.utils import DatabaseError, InternalError + +from pulpcore.app.entrypoint import PulpApiWorker + +_mock_connection = patch("pulpcore.app.entrypoint.connection") + + +def _make_read_only_error(): + cause = Exception("cannot execute UPDATE in a read-only transaction") + cause.pgcode = "25006" + exc = InternalError("read-only") + exc.__cause__ = cause + return exc + + +def _make_worker(): + with patch.object(PulpApiWorker, "__init__", lambda self, *a, **kw: None): + worker = PulpApiWorker() + worker.name = "test-worker@host" + worker.versions = {"core": "3.115.0"} + worker.app_status = MagicMock() + worker.beat_msg = "heartbeat ok" + worker.fail_beat_msg = "heartbeat failed" + return worker + + +@_mock_connection +class TestHeartbeatReadOnly: + def test_heartbeat_skips_on_read_only(self, _conn): + worker = _make_worker() + read_only_exc = _make_read_only_error() + worker.app_status.save_heartbeat.side_effect = [read_only_exc, read_only_exc] + + worker.heartbeat() + + assert worker.app_status.save_heartbeat.call_count == 2 + + def test_heartbeat_raises_on_real_db_error(self, _conn): + worker = _make_worker() + db_error = DatabaseError("connection lost") + worker.app_status.save_heartbeat.side_effect = [db_error, db_error] + + with pytest.raises(DatabaseError): + worker.heartbeat() + + def test_heartbeat_retries_create_when_app_status_none(self, _conn): + worker = _make_worker() + worker.app_status = None + read_only_exc = _make_read_only_error() + + with patch("pulpcore.app.models.AppStatus") as mock_cls: + mock_cls.objects.create.side_effect = read_only_exc + worker.heartbeat() + + assert worker.app_status is None + + def test_heartbeat_creates_app_status_when_db_recovers(self, _conn): + worker = _make_worker() + worker.app_status = None + mock_status = MagicMock() + + with patch("pulpcore.app.models.AppStatus") as mock_cls: + mock_cls.objects.create.return_value = mock_status + worker.heartbeat() + + assert worker.app_status is mock_status + mock_status.save_heartbeat.assert_called_once() + + +class TestRunCleanup: + def test_run_finally_tolerates_db_error(self): + worker = _make_worker() + worker.app_status.delete.side_effect = DatabaseError("read-only") + + with patch.object(PulpApiWorker.__bases__[0], "run"): + worker.run() + + worker.app_status.delete.assert_called_once() + + def test_run_finally_skips_when_no_app_status(self): + worker = _make_worker() + worker.app_status = None + + with patch.object(PulpApiWorker.__bases__[0], "run"): + worker.run() diff --git a/pulpcore/tests/unit/test_util.py b/pulpcore/tests/unit/test_util.py index 6b2943ceae3..6f9172b3eb4 100644 --- a/pulpcore/tests/unit/test_util.py +++ b/pulpcore/tests/unit/test_util.py @@ -222,3 +222,54 @@ def test_large_chunk_rotation(self) -> None: self.assertEqual(len(writer.results), 4) self.assertEqual((self.test_dir / "export.tar.0003").stat().st_size, 5) + + +class TestIsReadOnlyDbError: + @staticmethod + def _make_cause(pgcode=None, sqlstate=None): + cause = Exception("db error") + if pgcode is not None: + cause.pgcode = pgcode + if sqlstate is not None: + cause.sqlstate = sqlstate + return cause + + def test_pgcode_25006(self): + from pulpcore.app.util import _is_read_only_db_error + + cause = self._make_cause(pgcode="25006") + exc = Exception("read-only") + exc.__cause__ = cause + assert _is_read_only_db_error(exc) is True + + def test_sqlstate_25006(self): + from pulpcore.app.util import _is_read_only_db_error + + cause = self._make_cause(sqlstate="25006") + exc = Exception("read-only") + exc.__cause__ = cause + assert _is_read_only_db_error(exc) is True + + def test_no_cause(self): + from pulpcore.app.util import _is_read_only_db_error + + exc = Exception("connection lost") + assert _is_read_only_db_error(exc) is False + + def test_different_pgcode(self): + from pulpcore.app.util import _is_read_only_db_error + + cause = self._make_cause(pgcode="08006") + exc = Exception("connection failure") + exc.__cause__ = cause + assert _is_read_only_db_error(exc) is False + + def test_nested_cause(self): + from pulpcore.app.util import _is_read_only_db_error + + inner = self._make_cause(pgcode="25006") + outer = Exception("wrapper") + outer.__cause__ = inner + exc = Exception("nested") + exc.__cause__ = outer + assert _is_read_only_db_error(exc) is True