Server IP : 66.29.132.122 / Your IP : 3.144.222.60 Web Server : LiteSpeed System : Linux business142.web-hosting.com 4.18.0-553.lve.el8.x86_64 #1 SMP Mon May 27 15:27:34 UTC 2024 x86_64 User : admazpex ( 531) PHP Version : 7.2.34 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/proc/self/root/proc/thread-self/root/proc/self/root/opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/plugins/ |
Upload File : |
import asyncio import time from contextlib import suppress from datetime import timedelta from logging import getLogger from typing import Union from defence360agent.contracts.messages import MessageType from defence360agent.contracts.plugins import MessageSource from defence360agent.subsys.backup_systems import ( get_current_backend, get_last_backup_timestamp, ) from defence360agent.subsys.persistent_state import load_state, save_state from defence360agent.utils import Scope, recurring_check logger = getLogger(__name__) SEND_INTERVAL = int(timedelta(hours=24).total_seconds()) RECURRING_CHECK_INTERVAL = 5 class BackupInfoSender(MessageSource): """Send user backup statistics to CH periodically""" SCOPE = Scope.IM360 async def create_source(self, loop, sink): self._loop = loop self._sink = sink self._send_event = asyncio.Event() self._last_send_timestamp = self.load_last_send_timestamp() self._check_task = self._loop.create_task( self._recurring_check_data_to_send() ) self._send_stat_task = self._loop.create_task( self._recurring_send_stat() ) async def shutdown(self): for task in [self._check_task, self._send_stat_task]: task.cancel() with suppress(asyncio.CancelledError): await task self.save_last_send_timestamp() @staticmethod def is_valid_timestamp(timestamp: Union[int, float]) -> bool: return isinstance(timestamp, (int, float)) and timestamp > 0 def save_last_send_timestamp(self, ts: Union[int, float] = None): timestamp = self._last_send_timestamp if ts is None else ts if not self.is_valid_timestamp(timestamp): logger.warning("Invalid timestamp: %s", timestamp) return save_state("BackupInfoSender", {"last_send_timestamp": timestamp}) def load_last_send_timestamp(self): timestamp = load_state("BackupInfoSender").get("last_send_timestamp") if not self.is_valid_timestamp(timestamp): logger.warning("Invalid timestamp loaded, resetting to 0") timestamp = 0 return timestamp @recurring_check(RECURRING_CHECK_INTERVAL) async def _recurring_check_data_to_send(self): if time.time() - self._last_send_timestamp >= SEND_INTERVAL: self._send_event.set() @recurring_check(0) async def _recurring_send_stat(self): await self._send_event.wait() try: await self._send_server_config() except Exception as e: logger.exception("Failed to collect backup info: %s", e) finally: # Ensure backup info is not sent too frequently, even after an error self._last_send_timestamp = time.time() self._send_event.clear() async def _send_server_config(self): confg_msg = MessageType.BackupInfo( backup_provider_type=get_current_backend(), last_backup_timestamp=await get_last_backup_timestamp(), ) await self._sink.process_message(confg_msg)