Feat/monitor log size

django_logging/management/commands/logs_size_audit.py

@@ -0,0 +1,122 @@
+import logging
+import os
+from typing import Any, Dict, Tuple
+
+from django.conf import settings
+from django.core.management.base import BaseCommand
+
+from django_logging.constants import DefaultLoggingSettings
+from django_logging.constants.config_types import LogDir
+from django_logging.handlers import EmailHandler
+from django_logging.management.commands.send_logs import Command as cmd
+from django_logging.utils.get_conf import (
+    get_log_dir_size_limit,
+    use_email_notifier_template,
+)
+from django_logging.utils.log_email_notifier.notifier import send_email_async
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Command to check the total size of the logs directory and send a warning
+    if it exceeds the limit.
+
+    This command calculates the total size of the log directory and sends an email notification
+    to the admin if the size exceeds the configured limit.
+
+    Attributes:
+        help (str): A brief description of the command's functionality.
+
+    """
+
+    help = "Check the total size of the logs directory and send a warning if it exceeds the limit"
+
+    def handle(self, *args: Tuple[Any], **kwargs: Dict[str, Any]) -> None:
+        """Handles the command execution.
+
+        Args:
+            *args: Positional arguments passed to the command.
+            **kwargs: Keyword arguments passed to the command.
+
+        """
+        default_settings = DefaultLoggingSettings()
+        log_dir: LogDir = settings.DJANGO_LOGGING.get(
+            "LOG_DIR", os.path.join(os.getcwd(), default_settings.log_dir)
+        )
+
+        # Check if log directory exists
+        if not os.path.exists(log_dir):
+            self.stdout.write(self.style.ERROR(f"Log directory not found: {log_dir}"))
+            logger.error("Log directory not found: %s", log_dir)
+            return
+
+        # pylint: disable=attribute-defined-outside-init
+        self.size_limit: int = get_log_dir_size_limit()
+
+        # Calculate the total size of the log directory
+        total_size = self.get_directory_size(log_dir)
+        total_size_mb = float(f"{total_size / (1024 * 1024):.2f}")
+
+        logger.info("Total log directory size: %s MB", total_size_mb)
+
+        if int(total_size_mb) >= self.size_limit:
+            cmd.validate_email_settings()
+            # Send warning email if total size exceeds the size limit
+            self.send_warning_email(total_size_mb)
+            self.stdout.write(self.style.SUCCESS("Warning email sent successfully."))
+            logger.info("Warning email sent successfully.")
+        else:
+            self.stdout.write(
+                self.style.SUCCESS(
+                    f"Log directory size is under the limit: {total_size_mb} MB"
+                )
+            )
+            logger.info("Log directory size is under the limit: %s MB", total_size_mb)
+
+    # pylint: disable=unused-variable
+    def get_directory_size(self, dir_path: str) -> int:
+        """Calculate the total size of all files in the directory.
+
+        Args:
+            dir_path (str): The path of the directory to calculate size for.
+
+        Returns:
+            int: The total size of the directory in bytes.
+
+        """
+        total_size = 0
+        for root, dirs, files in os.walk(dir_path):
+            for file in files:
+                file_path = os.path.join(root, file)
+                total_size += os.path.getsize(file_path)
+
+        return total_size
+
+    def send_warning_email(self, total_size_mb: float) -> None:
+        """Send an email warning to the admin about the log directory size.
+
+        Args:
+            total_size_mb (int): The total size of the log directory in Megabytes.
+
+        """
+
+        subject = "Logs Directory Size Warning"
+        recipient_list = [settings.ADMIN_EMAIL]
+        message = (
+            f"The size of the log files has exceeded {self.size_limit} MB.\n\n"
+            f"Current size: {total_size_mb} MB\n"
+        )
+
+        email_body = message
+
+        if use_email_notifier_template():
+            email_body = EmailHandler.render_template(message)
+
+        send_email_async(
+            subject=subject, recipient_list=recipient_list, body=email_body
+        )
+        logger.info(
+            "Email has been sent to %s regarding log size warning.",
+            settings.ADMIN_EMAIL,
+        )

django_logging/management/commands/send_logs.py

@@ -48,11 +48,12 @@ def handle(self, *args: Tuple, **kwargs: Dict) -> None:
             kwargs (dict): Keyword arguments.
 
         """
-        email = kwargs["email"]
+        email: str = kwargs["email"]  # type: ignore
 
         default_settings = DefaultLoggingSettings()
+        log_settings = getattr(settings, "DJANGO_LOGGING", {})
 
-        log_dir: LogDir = settings.DJANGO_LOGGING.get(
+        log_dir: LogDir = log_settings.get(
             "LOG_DIR", os.path.join(os.getcwd(), default_settings.log_dir)
         )
 
@@ -97,15 +98,16 @@ def handle(self, *args: Tuple, **kwargs: Dict) -> None:
                 os.remove(zip_path)
                 logger.info("Temporary zip file cleaned up successfully.")
 
-    def validate_email_settings(self) -> None:
+    @staticmethod
+    def validate_email_settings(require_admin_email: bool = False) -> None:
         """Check if all required email settings are present in the settings
         file.
 
         Raises ImproperlyConfigured if any of the required email
         settings are missing.
 
         """
-        errors = check_email_settings(require_admin_email=False)
+        errors = check_email_settings(require_admin_email=require_admin_email)
         if errors:
             logger.error(errors)
             raise ImproperlyConfigured(errors)

django_logging/middleware/__init__.py

@@ -1 +1,2 @@
+from .monitor_log_size import MonitorLogSizeMiddleware
 from .request_middleware import RequestLogMiddleware

django_logging/middleware/monitor_log_size.py

@@ -0,0 +1,114 @@
+import logging
+from datetime import timedelta
+from typing import Awaitable, Callable, Union
+
+from asgiref.sync import sync_to_async
+from django.core.cache import cache
+from django.core.management import call_command
+from django.http import HttpRequest, HttpResponseBase
+from django.utils.timezone import now
+
+from django_logging.middleware.base import (  # pylint: disable=E0401, E0611
+    BaseMiddleware,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class MonitorLogSizeMiddleware(BaseMiddleware):
+    """Middleware that monitors the size of the log directory in both
+    synchronous and asynchronous modes.
+
+    This middleware checks if a week has passed since the last log size audit. If so, it runs
+    the 'logs_size_audit' management command, which checks the log directory's total size.
+    If the size exceeds a configured limit, a warning email is sent to the admin.
+
+    Attributes:
+    ----------
+        get_response (Callable[[HttpRequest], Union[HttpResponseBase, Awaitable[HttpResponseBase]]]):
+            The next middleware or view to be called.
+
+    """
+
+    # pylint: disable=useless-parent-delegation
+    def __init__(
+        self,
+        get_response: Callable[
+            [HttpRequest], Union[HttpResponseBase, Awaitable[HttpResponseBase]]
+        ],
+    ) -> None:
+        """Initializes the middleware with the provided get_response callable.
+
+        Args:
+        ----
+            get_response (callable): The next middleware or view to be called in the chain.
+
+        """
+        super().__init__(get_response)
+
+    def __sync_call__(self, request: HttpRequest) -> HttpResponseBase:
+        """Synchronous request processing.
+
+        Args:
+        ----
+            request (HttpRequest): The current HTTP request being processed.
+
+        Returns:
+        -------
+            HttpResponseBase: The HTTP response returned by the next middleware or view.
+
+        """
+        if self.should_run_task():
+            self.run_log_size_check()
+            cache.set("last_run_logs_size_audit", now(), timeout=None)
+
+        return self.get_response(request)
+
+    async def __acall__(self, request: HttpRequest) -> HttpResponseBase:
+        """Asynchronous request processing.
+
+        Args:
+        ----
+            request (HttpRequest): The current HTTP request being processed.
+
+        Returns:
+        -------
+            HttpResponseBase: The HTTP response returned by the next middleware or view.
+
+        """
+        if await sync_to_async(self.should_run_task)():
+            await sync_to_async(self.run_log_size_check)()
+            await sync_to_async(cache.set)(
+                "last_run_logs_size_audit", now(), timeout=None
+            )
+
+        return await self.get_response(request)
+
+    @staticmethod
+    def should_run_task() -> bool:
+        """Determines if a week has passed since the last log size audit.
+
+        Returns:
+        -------
+            bool: True if a week has passed since the last audit, False otherwise.
+
+        """
+        last_run = cache.get("last_run_logs_size_audit")
+        if last_run is None or now() - last_run > timedelta(weeks=1):
+            return True
+
+        return False
+
+    def run_log_size_check(self) -> None:
+        """Runs the 'logs_size_audit' management command to check the log
+        directory size.
+
+        If an error occurs during the execution of the command, it is
+        logged.
+
+        """
+        logger.info("Running 'logs_size_audit' command...")
+        try:
+            call_command("logs_size_audit")
+        except Exception as e:  # pylint: disable=W0718
+            logger.error("Error running 'logs_size_audit' command: %s", e)