File Manager

X7ROOT File Manager

Current Path: /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins

opt / imunify360 / venv / lib / python3.11 / site-packages / defence360agent / plugins /

?? ..
?? __init__.py(0 B)
?? __pycache__
?? accumulate.py(3.63 KB)
?? analyst_cleanup_update.py(5.53 KB)
?? backup_info_sender.py(3.06 KB)
?? cagefs.py(5.17 KB)
?? checkpoint.py(1.23 KB)
?? client.py(10.6 KB)
?? config_merger.py(828 B)
?? config_watcher.py(1.89 KB)
?? event_hook_executor.py(777 B)
?? event_monitor.py(3.32 KB)
?? event_monitor_message_processor.py(6.33 KB)
?? files_recurring_update.py(1.09 KB)
?? icontact_sender.py(4.42 KB)
?? idle_time_out.py(1.21 KB)
?? lve_utils_install.py(1.58 KB)
?? myimunify.py(1.98 KB)
?? ping.py(536 B)
?? send_domain_list.py(2.78 KB)
?? send_server_config.py(10.96 KB)

Editing: checkpoint.py

from defence360agent.contracts.plugins import MessageSink
from defence360agent.model.instance import db
from defence360agent.utils import recurring_check

class Checkpoint(MessageSink):
    """
    Checkpoint imunify360.db periodically to limit unexpected WAL file growing.
    """

ONE_DAY = 24 * 60 * 60

def __init__(self, *, checkpoint_period=ONE_DAY, db=db):
        self._checkpoint_period = checkpoint_period
        self._db = db
        self._task = None

async def create_sink(self, loop):
        self._loop = loop
        self._task = self._loop.create_task(
            recurring_check(self._checkpoint_period)(self._checkpoint)()
        )

async def shutdown(self):
        task, self._task = self._task, None  # avoid cancelling twice
        if task is None or task.cancelled():
            return
        task.cancel()
        # CancelledError is handled by @recurring_check():
        await task

async def _checkpoint(self):
        # 1. may not shrink database wal file in case of this command will be
        # during external read process took place
        # 2. returning immediately without result if database
        # has concurrent transaction
        self._db.execute_sql("PRAGMA wal_checkpoint(TRUNCATE)")

X7ROOT File Manager

Editing: checkpoint.py

Upload File

Create Folder