From 029d22ff9c83e3532689a66aae8ac7072e74f94d Mon Sep 17 00:00:00 2001 From: pablorevilla-meshtastic Date: Thu, 4 Jun 2026 17:14:43 -0700 Subject: [PATCH] report of cleanup status on health page --- .gitignore | 1 + docs/API_Documentation.md | 43 ++++++++++++++++++ meshview/mqtt_store.py | 19 ++++++-- meshview/web_api/api.py | 92 ++++++++++++++++++++++++++++++++++++++ startdb.py | 94 ++++++++++++++++++++++++++++++++++++++- 5 files changed, 243 insertions(+), 6 deletions(-) diff --git a/.gitignore b/.gitignore index 7a94de8..d4f6cdd 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,7 @@ meshview-web.pid /table_details.py config.ini *.log +*.status.json # Screenshots screenshots/* diff --git a/docs/API_Documentation.md b/docs/API_Documentation.md index bc8e10a..564c135 100644 --- a/docs/API_Documentation.md +++ b/docs/API_Documentation.md @@ -368,6 +368,49 @@ Response Example "timestamp": "2025-07-22T12:45:00+00:00", "version": "3.0.3", "git_revision": "abc1234", + "cleanup": { + "enabled": true, + "days_to_keep": 14, + "scheduled_time": "02:00", + "vacuum": false, + "status": "ok", + "status_file": "dbcleanup.status.json", + "last_run": { + "status": "ok", + "started_at": "2026-06-04T09:00:00+00:00", + "completed_at": "2026-06-04T09:00:03+00:00", + "cutoff_at": "2026-05-21T09:00:00+00:00", + "days_to_keep": 14, + "vacuum_requested": false, + "vacuum_completed": false, + "rows_deleted": { + "packet": 1200, + "packet_seen": 3400, + "traceroute": 42, + "node": 3 + }, + "error": null + } + }, + "backup": { + "enabled": true, + "backup_dir": "./backups", + "scheduled_time": "02:00", + "status": "ok", + "status_file": "dbbackup.status.json", + "last_run": { + "status": "ok", + "started_at": "2026-06-04T09:00:00+00:00", + "completed_at": "2026-06-04T09:00:02+00:00", + "backup_dir": "./backups", + "database_path": "packets.db", + "backup_file": "backups/packets_backup_20260604_090000.db.gz", + "original_size_bytes": 12939444, + "compressed_size_bytes": 4211560, + "compression_percent": 67.5, + "error": null + } + }, "database": "connected", "database_size": "12.34 MB", "database_size_bytes": 12939444 diff --git a/meshview/mqtt_store.py b/meshview/mqtt_store.py index 825de1f..6bd9bab 100644 --- a/meshview/mqtt_store.py +++ b/meshview/mqtt_store.py @@ -17,6 +17,15 @@ from meshview.models import DailySnapshot, Node, Packet, PacketSeen, Traceroute logger = logging.getLogger(__name__) MQTT_GATEWAY_CACHE: set[int] = set() +UNKNOWN_NODE_NAME = "unknown name" +NODE_NAME_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]") + + +def normalize_node_name(name: str) -> str: + name = name.strip() + if not name or NODE_NAME_CONTROL_CHARS.search(name): + return UNKNOWN_NODE_NAME + return name async def capture_daily_snapshot() -> None: @@ -268,11 +277,13 @@ async def process_envelope(topic, env): ).scalar_one_or_none() now_us = int(time.time() * 1_000_000) + long_name = normalize_node_name(user.long_name) + short_name = normalize_node_name(user.short_name) if node: node.node_id = node_id - node.long_name = user.long_name - node.short_name = user.short_name + node.long_name = long_name + node.short_name = short_name node.hw_model = hw_model node.role = role node.channel = env.channel_id @@ -283,8 +294,8 @@ async def process_envelope(topic, env): node = Node( id=user.id, node_id=node_id, - long_name=user.long_name, - short_name=user.short_name, + long_name=long_name, + short_name=short_name, hw_model=hw_model, role=role, channel=env.channel_id, diff --git a/meshview/web_api/api.py b/meshview/web_api/api.py index 1a8956d..71e5c99 100644 --- a/meshview/web_api/api.py +++ b/meshview/web_api/api.py @@ -38,6 +38,96 @@ _LANG_CACHE = {} routes = web.RouteTableDef() +def _config_bool(section: str, key: str, default: bool = False) -> bool: + return str(CONFIG.get(section, {}).get(key, default)).lower() in ("1", "true", "yes", "on") + + +def _config_int(section: str, key: str, default: int = 0) -> int: + try: + return int(CONFIG.get(section, {}).get(key, default)) + except (TypeError, ValueError): + return default + + +def _config_str(section: str, key: str, default: str = "") -> str: + value = CONFIG.get(section, {}).get(key, default) + return str(value) if value is not None else default + + +def _cleanup_status_file() -> str: + cleanup_logfile = CONFIG.get("logging", {}).get("db_cleanup_logfile", "dbcleanup.log") + path_without_extension, _ = os.path.splitext(cleanup_logfile) + return f"{path_without_extension}.status.json" + + +def _backup_status_file() -> str: + cleanup_logfile = CONFIG.get("logging", {}).get("db_cleanup_logfile", "dbcleanup.log") + return os.path.join(os.path.dirname(cleanup_logfile), "dbbackup.status.json") + + +def _get_cleanup_health() -> dict: + cleanup_health = { + "enabled": _config_bool("cleanup", "enabled", False), + "days_to_keep": _config_int("cleanup", "days_to_keep", 14), + "scheduled_time": ( + f"{_config_int('cleanup', 'hour', 2):02d}:{_config_int('cleanup', 'minute', 0):02d}" + ), + "vacuum": _config_bool("cleanup", "vacuum", False), + "status": "disabled", + } + + if cleanup_health["enabled"]: + cleanup_health["status"] = "unknown" + + status_file = _cleanup_status_file() + cleanup_health["status_file"] = status_file + if not os.path.exists(status_file): + return cleanup_health + + try: + with open(status_file, encoding="utf-8") as f: + last_run = json.load(f) + except Exception as e: + cleanup_health["status"] = "error" + cleanup_health["error"] = f"Unable to read cleanup status: {e}" + return cleanup_health + + cleanup_health["status"] = last_run.get("status", cleanup_health["status"]) + cleanup_health["last_run"] = last_run + return cleanup_health + + +def _get_backup_health() -> dict: + backup_hour = _config_int("cleanup", "backup_hour", _config_int("cleanup", "hour", 2)) + backup_minute = _config_int("cleanup", "backup_minute", _config_int("cleanup", "minute", 0)) + backup_health = { + "enabled": _config_bool("cleanup", "backup_enabled", False), + "backup_dir": _config_str("cleanup", "backup_dir", "./backups"), + "scheduled_time": f"{backup_hour:02d}:{backup_minute:02d}", + "status": "disabled", + } + + if backup_health["enabled"]: + backup_health["status"] = "unknown" + + status_file = _backup_status_file() + backup_health["status_file"] = status_file + if not os.path.exists(status_file): + return backup_health + + try: + with open(status_file, encoding="utf-8") as f: + last_run = json.load(f) + except Exception as e: + backup_health["status"] = "error" + backup_health["error"] = f"Unable to read backup status: {e}" + return backup_health + + backup_health["status"] = last_run.get("status", backup_health["status"]) + backup_health["last_run"] = last_run + return backup_health + + def _haversine_km(lat1, lon1, lat2, lon2): r = 6371.0 phi1 = math.radians(lat1) @@ -720,6 +810,8 @@ async def health_check(request): "timestamp": datetime.datetime.now(datetime.UTC).isoformat(), "version": __version__, "git_revision": _git_revision_short, + "cleanup": _get_cleanup_health(), + "backup": _get_backup_health(), } # Check database connectivity diff --git a/startdb.py b/startdb.py index 1ef461d..6c08c38 100644 --- a/startdb.py +++ b/startdb.py @@ -33,6 +33,8 @@ file_handler.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s') file_handler.setFormatter(formatter) cleanup_logger.addHandler(file_handler) +cleanup_status_file = str(Path(cleanup_logfile).with_suffix(".status.json")) +backup_status_file = str(Path(cleanup_logfile).with_name("dbbackup.status.json")) # ------------------------- @@ -49,6 +51,32 @@ def get_int(config, section, key, default=0): return default +def _rowcount(value): + return value if value is not None and value >= 0 else None + + +def write_cleanup_status(status: dict) -> None: + status_path = Path(cleanup_status_file) + tmp_path = status_path.with_suffix(f"{status_path.suffix}.tmp") + try: + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(status, f, indent=2, sort_keys=True) + tmp_path.replace(status_path) + except Exception as e: + cleanup_logger.warning(f"Failed to write cleanup status file: {e}") + + +def write_backup_status(status: dict) -> None: + status_path = Path(backup_status_file) + tmp_path = status_path.with_suffix(f"{status_path.suffix}.tmp") + try: + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(status, f, indent=2, sort_keys=True) + tmp_path.replace(status_path) + except Exception as e: + cleanup_logger.warning(f"Failed to write backup status file: {e}") + + # ------------------------- # Shared DB lock # ------------------------- @@ -66,19 +94,40 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None: database_url: SQLAlchemy connection string backup_dir: Directory to store backups (default: current directory) """ + backup_status = { + "status": "running", + "started_at": datetime.datetime.now(datetime.UTC).isoformat(), + "completed_at": None, + "backup_dir": backup_dir, + "database_path": None, + "backup_file": None, + "original_size_bytes": None, + "compressed_size_bytes": None, + "compression_percent": None, + "error": None, + } + write_backup_status(backup_status) + try: url = make_url(database_url) if not url.drivername.startswith("sqlite"): cleanup_logger.warning("Backup only supported for SQLite databases") + backup_status["status"] = "unsupported" + backup_status["error"] = "Backup only supported for SQLite databases" return if not url.database or url.database == ":memory:": cleanup_logger.error("Could not extract database path from connection string") + backup_status["status"] = "error" + backup_status["error"] = "Could not extract database path from connection string" return db_file = Path(url.database) + backup_status["database_path"] = str(db_file) if not db_file.exists(): cleanup_logger.error(f"Database file not found: {db_file}") + backup_status["status"] = "error" + backup_status["error"] = f"Database file not found: {db_file}" return # Create backup directory if it doesn't exist @@ -89,6 +138,7 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None: timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"{db_file.stem}_backup_{timestamp}.db.gz" backup_file = backup_path / backup_filename + backup_status["backup_file"] = str(backup_file) cleanup_logger.info(f"Creating backup: {backup_file}") @@ -98,9 +148,15 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None: shutil.copyfileobj(f_in, f_out) # Get file sizes for logging - original_size = db_file.stat().st_size / (1024 * 1024) # MB - compressed_size = backup_file.stat().st_size / (1024 * 1024) # MB + original_size_bytes = db_file.stat().st_size + compressed_size_bytes = backup_file.stat().st_size + original_size = original_size_bytes / (1024 * 1024) # MB + compressed_size = compressed_size_bytes / (1024 * 1024) # MB compression_ratio = (1 - compressed_size / original_size) * 100 if original_size > 0 else 0 + backup_status["original_size_bytes"] = original_size_bytes + backup_status["compressed_size_bytes"] = compressed_size_bytes + backup_status["compression_percent"] = round(compression_ratio, 1) + backup_status["status"] = "ok" cleanup_logger.info( f"Backup created successfully: {backup_file.name} " @@ -110,6 +166,11 @@ async def backup_database(database_url: str, backup_dir: str = ".") -> None: except Exception as e: cleanup_logger.error(f"Error creating database backup: {e}") + backup_status["status"] = "error" + backup_status["error"] = str(e) + finally: + backup_status["completed_at"] = datetime.datetime.now(datetime.UTC).isoformat() + write_backup_status(backup_status) # ------------------------- @@ -179,6 +240,24 @@ async def daily_cleanup_at( ).replace(tzinfo=None) cutoff_us = int(cutoff_dt.timestamp() * 1_000_000) cleanup_logger.info(f"Running cleanup for records older than {cutoff_dt.isoformat()}...") + rows_deleted = { + "packet": None, + "packet_seen": None, + "traceroute": None, + "node": None, + } + cleanup_status = { + "status": "running", + "started_at": datetime.datetime.now(datetime.UTC).isoformat(), + "completed_at": None, + "cutoff_at": cutoff_dt.replace(tzinfo=datetime.UTC).isoformat(), + "days_to_keep": days_to_keep, + "vacuum_requested": vacuum_db, + "vacuum_completed": False, + "rows_deleted": rows_deleted, + "error": None, + } + write_cleanup_status(cleanup_status) try: async with db_lock: # Pause ingestion @@ -191,6 +270,7 @@ async def daily_cleanup_at( result = await session.execute( delete(models.Packet).where(models.Packet.import_time_us < cutoff_us) ) + rows_deleted["packet"] = _rowcount(result.rowcount) cleanup_logger.info(f"Deleted {result.rowcount} rows from Packet") # ------------------------- @@ -201,6 +281,7 @@ async def daily_cleanup_at( models.PacketSeen.import_time_us < cutoff_us ) ) + rows_deleted["packet_seen"] = _rowcount(result.rowcount) cleanup_logger.info(f"Deleted {result.rowcount} rows from PacketSeen") # ------------------------- @@ -211,6 +292,7 @@ async def daily_cleanup_at( models.Traceroute.import_time_us < cutoff_us ) ) + rows_deleted["traceroute"] = _rowcount(result.rowcount) cleanup_logger.info(f"Deleted {result.rowcount} rows from Traceroute") # ------------------------- @@ -219,6 +301,7 @@ async def daily_cleanup_at( result = await session.execute( delete(models.Node).where(models.Node.last_seen_us < cutoff_us) ) + rows_deleted["node"] = _rowcount(result.rowcount) cleanup_logger.info(f"Deleted {result.rowcount} rows from Node") await session.commit() @@ -228,14 +311,21 @@ async def daily_cleanup_at( async with mqtt_database.engine.begin() as conn: await conn.exec_driver_sql("VACUUM;") cleanup_logger.info("VACUUM completed.") + cleanup_status["vacuum_completed"] = True elif vacuum_db: cleanup_logger.info("VACUUM skipped (not supported for this database).") cleanup_logger.info("Cleanup completed successfully.") cleanup_logger.info("Ingestion resumed after cleanup.") + cleanup_status["status"] = "ok" except Exception as e: cleanup_logger.error(f"Error during cleanup: {e}") + cleanup_status["status"] = "error" + cleanup_status["error"] = str(e) + finally: + cleanup_status["completed_at"] = datetime.datetime.now(datetime.UTC).isoformat() + write_cleanup_status(cleanup_status) # -------------------------