Merge pull request #136 from aiven/kathiabarahona/disable_backups_until

Support backup intervals greater than 24 hours

#136

Author: kmichel-aiven

myhoard/controller.py

@@ -851,6 +851,17 @@ def _create_restore_coordinator_if_missing(self):
         else:
             self.log.info("Newly created restore coordinator is already in completed state")
 
+    def _previous_normalized_backup_timestamp(self) -> Optional[str]:
+        normalized_backup_times = [
+            stream.state["normalized_backup_time"]
+            for stream in self.backup_streams
+            if stream.state["normalized_backup_time"]
+        ]
+
+        if not normalized_backup_times:
+            return None
+        return max(normalized_backup_times)
+
     def _current_normalized_backup_timestamp(self) -> str:
         """Returns the closest historical backup time that current time matches to (or current time if it matches).
         E.g. if backup hour is 13, backup minute is 50, current time is 15:40 and backup interval is 60 minutes,
@@ -861,11 +872,19 @@ def _current_normalized_backup_timestamp(self) -> str:
         backup_interval_minutes = self.backup_settings["backup_interval_minutes"]
         backup_hour = self.backup_settings["backup_hour"]
         backup_minute = self.backup_settings["backup_minute"]
-        if normalized.hour < backup_hour or (normalized.hour == backup_hour and normalized.minute < backup_minute):
-            normalized = normalized - datetime.timedelta(days=1)
-        normalized = normalized.replace(hour=backup_hour, minute=backup_minute, second=0, microsecond=0)
-        while normalized + datetime.timedelta(minutes=backup_interval_minutes) < now:
+
+        previous_normalized = self._previous_normalized_backup_timestamp()
+        # in case we have a previous backup time, we can use it as starting point
+        if previous_normalized:
+            normalized = datetime.datetime.fromisoformat(previous_normalized)
+        else:
+            if normalized.hour < backup_hour or (normalized.hour == backup_hour and normalized.minute < backup_minute):
+                normalized = normalized - datetime.timedelta(days=1)
+            normalized = normalized.replace(hour=backup_hour, minute=backup_minute, second=0, microsecond=0)
+
+        while normalized + datetime.timedelta(minutes=backup_interval_minutes) <= now:
             normalized = normalized + datetime.timedelta(minutes=backup_interval_minutes)
+
         return normalized.isoformat()
 
     def _determine_unapplied_remote_binlogs(self, stream):
@@ -1177,23 +1196,14 @@ def _mark_failed_restore_backup_as_broken(self) -> None:
 
     def _mark_periodic_backup_requested_if_interval_exceeded(self):
         normalized_backup_time = self._current_normalized_backup_timestamp()
-        most_recent_scheduled = None
-        last_normalized_backup_time = None
-        if self.backup_streams:
-            normalized_backup_times = [
-                stream.state["normalized_backup_time"]
-                for stream in self.backup_streams
-                if stream.state["normalized_backup_time"]
-            ]
-            if normalized_backup_times:
-                last_normalized_backup_time = max(normalized_backup_times)
-            scheduled_streams = [
-                stream
-                for stream in self.backup_streams
-                if stream.state["backup_reason"] == BackupStream.BackupReason.scheduled
-            ]
-            if scheduled_streams:
-                most_recent_scheduled = max(stream.created_at for stream in scheduled_streams)
+        last_normalized_backup_time = self._previous_normalized_backup_timestamp()
+
+        scheduled_streams_created_at = [
+            stream.created_at
+            for stream in self.backup_streams
+            if stream.state["backup_reason"] == BackupStream.BackupReason.scheduled
+        ]
+        most_recent_scheduled = max(scheduled_streams_created_at) if scheduled_streams_created_at else None
 
         # Don't create new backup unless at least half of interval has elapsed since scheduled last backup. Otherwise
         # we would end up creating a new backup each time when backup hour/minute changes, which is typically undesired.

requirements.dev.txt

@@ -17,5 +17,6 @@ pytest-mock
 pytest-timeout
 pytest-xdist
 responses
+time-machine
 types-PyMySQL
 types-requests

test/test_controller.py

@@ -17,6 +17,7 @@
 from unittest.mock import MagicMock, patch
 
 import contextlib
+import datetime
 import os
 import pytest
 import random
@@ -1287,6 +1288,75 @@ def test_collect_binlogs_to_purge():
     log.info.assert_any_call("Binlog %s has been replicated to all servers, purging", 3)
 
 
+def test_periodic_backup_based_on_exceeded_intervals(time_machine, master_controller) -> None:
+    # pylint: disable=protected-access
+    time_machine.move_to("2023-01-02T18:00:00")
+
+    # By default backup_hour = 3, backup_interval_minutes = 1440
+    m_controller, master = master_controller
+
+    m_controller.switch_to_active_mode()
+    m_controller.start()
+
+    def streaming_binlogs(controller: Controller, expected_completed_backups: int):
+        assert controller.backup_streams
+        assert controller.backup_streams[0].active_phase == BackupStream.ActivePhase.binlog
+
+        complete_backups = [backup for backup in controller.state["backups"] if backup["completed_at"]]
+        assert len(complete_backups) == expected_completed_backups
+
+    def flush_binlogs():
+        with mysql_cursor(**master.connect_options) as cursor:
+            cursor.execute("FLUSH BINARY LOGS")
+
+    # write some data for the first backup
+    flush_binlogs()
+    # first backup for 2023-01-02 3:00 should be generated at 6pm (time we started the service)
+    while_asserts(lambda: streaming_binlogs(m_controller, 1), timeout=10)
+
+    # generate more data for second backup
+    flush_binlogs()
+
+    # second backup ideally should be done at 2023-01-03 03:00, but following the
+    # "half of interval" rule (at least 12 hours of difference between most recent scheduled and current time)
+    # therefore, it should be actually scheduled at 2023-01-03 06:00
+    time_machine.move_to("2023-01-03T03:00:00+00:00")
+
+    expected_normalized_time = datetime.datetime(2023, 1, 3, 3, tzinfo=datetime.timezone.utc)
+    assert m_controller._current_normalized_backup_timestamp() == expected_normalized_time.isoformat()
+
+    # no new backup should be scheduled
+    time.sleep(1)
+    min_created_at = datetime.datetime(2023, 1, 3, 3, tzinfo=datetime.timezone.utc).timestamp()
+    assert not any(bs.created_at >= min_created_at for bs in m_controller.backup_streams)
+
+    time_machine.move_to("2023-01-03T06:00:00+00:00")
+    while_asserts(lambda: streaming_binlogs(m_controller, 2), timeout=10)
+
+    # generate more data for third backup
+    flush_binlogs()
+
+    # After second backup, the next scheduled one should be at 2023-02-04 03:00:00, but let's change
+    # backup_interval_minutes to 2880 (48 hours)
+    m_controller.backup_settings["backup_interval_minutes"] = 2880
+
+    time_machine.move_to("2023-01-04T06:00:00+00:00")
+
+    expected_normalized_time = datetime.datetime(2023, 1, 3, 3, tzinfo=datetime.timezone.utc)
+    assert m_controller._current_normalized_backup_timestamp() == expected_normalized_time.isoformat()
+
+    time.sleep(1)
+    # no new backup should be scheduled
+    min_created_at = datetime.datetime(2023, 1, 4, 3, tzinfo=datetime.timezone.utc).timestamp()
+    assert not any(bs.created_at >= min_created_at for bs in m_controller.backup_streams)
+
+    time_machine.move_to("2023-01-05T03:00:00+00:00")
+    expected_normalized_time = datetime.datetime(2023, 1, 5, 3, tzinfo=datetime.timezone.utc)
+
+    assert m_controller._current_normalized_backup_timestamp() == expected_normalized_time.isoformat()
+    while_asserts(lambda: streaming_binlogs(m_controller, 3), timeout=10)
+
+
 @patch.object(RestoreCoordinator, "MAX_BASEBACKUP_ERRORS", 2)
 @patch.object(BasebackupRestoreOperation, "restore_backup", side_effect=Exception("failed restoring basebackup"))
 def test_backup_marked_as_broken_after_failed_restoration(