Add database backups and automatic restore
Daemon job "backup" (online copy, quick_check, rotation), restore of the newest valid copy when the database cannot be opened, change journal reset after restore, last_restore in /health, docs and tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
ayurishchevandClaude Sonnet 5 committed
1 parent
c99665542a
commit
02f7b49e12
10 files changed
+294
-21
No files matched your search
@@ -1,16 +1,20 @@
|
||||
"""Хранение собранных адресов в SQLite (ripe.db). Конфигурация и heartbeat остаются в JSON."""
|
||||
import datetime
|
||||
import glob
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sqlite3
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
|
||||
from storage import StorageError, load_json
|
||||
from storage import StorageError, file_lock, load_json, save_json_atomic
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SCHEMA_VERSION = 2
|
||||
# После отката к копии номера журнала сдвигаются на это число: курсоры, выданные до отката, становятся недействительными
|
||||
RESTORE_JOURNAL_JUMP = 1_000_000
|
||||
|
||||
_UPSERT = ("INSERT INTO addresses (kind, source, value, first_seen, last_seen) VALUES (?, ?, ?, ?, ?) "
|
||||
"ON CONFLICT (kind, source, value) DO UPDATE SET last_seen = excluded.last_seen")
|
||||
@@ -29,28 +33,148 @@ def _quarantine(path):
|
||||
return f"{path}.corrupt-{stamp}"
|
||||
|
||||
|
||||
def connect(path=None):
|
||||
"""Открывает базу, создаёт схему и однократно импортирует старые data.json / fqdn_data.json."""
|
||||
import cidr_collector as cc # пути читаем при вызове (их подменяют тесты); импорт отложен из-за цикла
|
||||
|
||||
path = path or cc.DB_FILE
|
||||
def _open(path, cc):
|
||||
"""Открывает базу и приводит схему к актуальной. Ошибки sqlite3 пробрасываются как есть."""
|
||||
conn = sqlite3.connect(path, timeout=5.0, isolation_level=None) # транзакциями управляем явно
|
||||
try:
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA synchronous=NORMAL")
|
||||
conn.execute("PRAGMA temp_store=MEMORY")
|
||||
_ensure_schema(conn, cc)
|
||||
except sqlite3.OperationalError as e: # например, database is locked: базу не трогаем
|
||||
except BaseException:
|
||||
conn.close()
|
||||
raise
|
||||
return conn
|
||||
|
||||
|
||||
def connect(path=None):
|
||||
"""Открывает базу, создаёт схему и однократно импортирует старые data.json / fqdn_data.json.
|
||||
|
||||
Повреждённую базу убирает в карантин и восстанавливает из последней исправной копии (см. _recover).
|
||||
"""
|
||||
import cidr_collector as cc # пути читаем при вызове (их подменяют тесты); импорт отложен из-за цикла
|
||||
|
||||
path = path or cc.DB_FILE
|
||||
try:
|
||||
return _open(path, cc)
|
||||
except sqlite3.OperationalError as e: # например, database is locked: базу не трогаем
|
||||
raise StorageError(f"Cannot open {path}: {e}") from e
|
||||
except sqlite3.DatabaseError as e: # not a database / malformed
|
||||
return _recover(path, e, cc)
|
||||
|
||||
|
||||
def list_backups(backup_dir):
|
||||
"""Копии от новых к старым (имя содержит время UTC, поэтому сортировка по имени = по времени)."""
|
||||
return sorted(glob.glob(os.path.join(backup_dir, "ripe-*.db")), reverse=True)
|
||||
|
||||
|
||||
def _quick_check_file(path):
|
||||
"""True, если файл открывается как база и проходит PRAGMA quick_check."""
|
||||
try:
|
||||
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5.0)
|
||||
try:
|
||||
return conn.execute("PRAGMA quick_check").fetchone()[0] == "ok"
|
||||
finally:
|
||||
conn.close()
|
||||
except sqlite3.Error:
|
||||
return False
|
||||
|
||||
|
||||
def _recover(path, error, cc):
|
||||
"""Порча базы при открытии: карантин и восстановление из последней исправной копии.
|
||||
|
||||
API и демон могут обнаружить порчу одновременно, поэтому всё выполняется под блокировкой,
|
||||
а внутри неё база проверяется повторно (другой процесс мог уже восстановить). Нет копий - StorageError.
|
||||
"""
|
||||
with file_lock(path + ".restore"):
|
||||
try:
|
||||
return _open(path, cc)
|
||||
except sqlite3.OperationalError as e:
|
||||
raise StorageError(f"Cannot open {path}: {e}") from e
|
||||
except sqlite3.DatabaseError:
|
||||
pass
|
||||
quarantine = _quarantine(path)
|
||||
logger.error("Corrupted database %s (%s); moved to %s", path, error, quarantine)
|
||||
for candidate in list_backups(cc.BACKUP_DIR):
|
||||
if not _quick_check_file(candidate):
|
||||
logger.warning("Backup %s failed the integrity check, skipping", candidate)
|
||||
continue
|
||||
try:
|
||||
conn = _restore_backup(candidate, path, cc)
|
||||
except (sqlite3.Error, OSError) as e:
|
||||
logger.warning("Cannot restore from %s: %s", candidate, e)
|
||||
continue
|
||||
logger.error("Database restored from backup %s; data collected after it is lost "
|
||||
"and will be gathered again by the collector", candidate)
|
||||
save_json_atomic(cc.RESTORE_FILE, {"at": datetime.datetime.now().isoformat(timespec="seconds"),
|
||||
"backup": candidate, "quarantine": quarantine, "error": str(error)})
|
||||
return conn
|
||||
raise StorageError(f"{path} is corrupted and no valid backup was found") from error
|
||||
|
||||
|
||||
def _restore_backup(candidate, path, cc):
|
||||
"""Кладёт копию на место базы и сбрасывает журнал изменений (его номера после точки копии уже выдавались)."""
|
||||
tmp = path + ".restore.tmp"
|
||||
shutil.copyfile(candidate, tmp)
|
||||
os.replace(tmp, path)
|
||||
conn = _open(path, cc)
|
||||
try:
|
||||
conn.execute("BEGIN IMMEDIATE")
|
||||
try:
|
||||
row = conn.execute("SELECT seq FROM sqlite_sequence WHERE name = 'changes'").fetchone()
|
||||
horizon = int(conn.execute("SELECT value FROM meta WHERE key = 'horizon_id'").fetchone()[0])
|
||||
new_seq = max(row[0] if row else 0, horizon) + RESTORE_JOURNAL_JUMP
|
||||
conn.execute("DELETE FROM changes")
|
||||
if row:
|
||||
conn.execute("UPDATE sqlite_sequence SET seq = ? WHERE name = 'changes'", (new_seq,))
|
||||
else:
|
||||
conn.execute("INSERT INTO sqlite_sequence (name, seq) VALUES ('changes', ?)", (new_seq,))
|
||||
conn.execute("UPDATE meta SET value = ? WHERE key = 'horizon_id'", (str(new_seq),))
|
||||
conn.execute(f"UPDATE meta SET value = {_TS} WHERE key = 'horizon_ts'")
|
||||
except BaseException:
|
||||
conn.execute("ROLLBACK")
|
||||
raise
|
||||
conn.execute("COMMIT")
|
||||
except BaseException:
|
||||
conn.close()
|
||||
backup = _quarantine(path)
|
||||
logger.error("Corrupted database %s (%s); moved to %s", path, e, backup)
|
||||
raise StorageError(f"{path} is corrupted") from e
|
||||
raise
|
||||
return conn
|
||||
|
||||
|
||||
def backup_database(conn, now, keep, backup_dir=None):
|
||||
"""Онлайн-копия базы с проверкой и ротацией. Возвращает путь копии.
|
||||
|
||||
Живая база и копия проверяются PRAGMA quick_check; при неудаче копия не появляется, старые не трогаются
|
||||
(исключение StorageError). Лишние старые копии (сверх keep) удаляются только после успешной новой.
|
||||
"""
|
||||
import cidr_collector as cc
|
||||
|
||||
backup_dir = backup_dir or cc.BACKUP_DIR
|
||||
if conn.execute("PRAGMA quick_check").fetchone()[0] != "ok":
|
||||
raise StorageError("Live database failed the integrity check, backup skipped")
|
||||
os.makedirs(backup_dir, exist_ok=True)
|
||||
for stale in glob.glob(os.path.join(backup_dir, "ripe-*.db.tmp")): # хвосты прерванной копии
|
||||
os.unlink(stale)
|
||||
|
||||
stamp = now.astimezone(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||||
path = os.path.join(backup_dir, f"ripe-{stamp}.db")
|
||||
tmp = path + ".tmp"
|
||||
dst = sqlite3.connect(tmp)
|
||||
try:
|
||||
conn.backup(dst)
|
||||
dst.execute("PRAGMA journal_mode=DELETE") # копия - один самодостаточный файл без -wal
|
||||
finally:
|
||||
dst.close()
|
||||
if not _quick_check_file(tmp):
|
||||
os.unlink(tmp)
|
||||
raise StorageError("Backup failed the integrity check and was discarded")
|
||||
os.replace(tmp, path)
|
||||
|
||||
for old in list_backups(backup_dir)[keep:]:
|
||||
os.unlink(old)
|
||||
return path
|
||||
|
||||
|
||||
@contextmanager
|
||||
def session():
|
||||
"""Соединение на время блока (закрывается по выходу)."""
|
||||
|
||||
Reference in new issue
Block a user