diff --git a/.gitignore b/.gitignore index 018ed8c..811e593 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,4 @@ nonpacks/media/ nonpacks/static/vanilla/entity_index.json *.log AGENTS/ +logs/ diff --git a/nonpacks/landing/tests.py b/nonpacks/landing/tests.py index 6b40777..47b7494 100644 --- a/nonpacks/landing/tests.py +++ b/nonpacks/landing/tests.py @@ -1,3 +1,8 @@ +import tempfile +import time +import unittest.mock +from pathlib import Path + from django.contrib.auth import get_user_model from django.core.cache import cache from django.test import TestCase @@ -26,7 +31,7 @@ class ServerStatsTests(TestCase): resp = self.client.get(reverse('landing:server_stats_api')) self.assertEqual(resp.status_code, 200) data = resp.json() - for key in ('app', 'system', 'content', 'recent', 'generated_at'): + for key in ('app', 'system', 'content', 'recent', 'traffic', 'generated_at'): self.assertIn(key, data) self.assertIn('users', data['content']) self.assertIn('versions', data['recent']) @@ -59,3 +64,33 @@ class ServerStatsTests(TestCase): self._authorized() resp = self.client.get(reverse('landing:server_stats')) self.assertEqual(resp.status_code, 302) + + +class TrafficStatsTests(TestCase): + def test_parses_access_log(self): + from landing import views + with tempfile.TemporaryDirectory() as tmp: + log = Path(tmp) / 'gunicorn-access.log' + now = time.strftime('%d/%b/%Y:%H:%M:%S %z', time.localtime()) + log.write_text(''.join([ + f'127.0.0.1 [{now}] "GET /home/ HTTP/1.1" 200 1200 0.004\n', + f'127.0.0.1 [{now}] "POST /api/uploads/ HTTP/1.1" 200 80 0.020\n', + f'127.0.0.1 [{now}] "GET /packs/test-pack/ HTTP/1.1" 404 300 0.001\n', + f'10.0.0.5 [{now}] "GET /api/stats/ HTTP/1.1" 500 50 0.150\n', + ])) + with unittest.mock.patch.object(views, 'GUNICORN_ACCESS_LOG', log): + stats = views._traffic_stats() + self.assertIsNotNone(stats) + self.assertEqual(stats['requests_total'], 4) + self.assertEqual(stats['requests_min'], 4) + self.assertEqual(stats['status'], {'2xx': 2, '3xx': 0, '4xx': 1, '5xx': 1}) + self.assertEqual(stats['max_ms'], 150.0) + self.assertEqual(len(stats['recent']), 4) + self.assertEqual(stats['recent'][0]['status'], 500) + + def test_missing_log_returns_none(self): + from landing import views + with tempfile.TemporaryDirectory() as tmp: + log = Path(tmp) / 'nope.log' + with unittest.mock.patch.object(views, 'GUNICORN_ACCESS_LOG', log): + self.assertIsNone(views._traffic_stats()) diff --git a/nonpacks/landing/views.py b/nonpacks/landing/views.py index 9a0c3f4..d6b6979 100644 --- a/nonpacks/landing/views.py +++ b/nonpacks/landing/views.py @@ -1,9 +1,13 @@ import hmac +import os import platform import random +import re import socket +import statistics import time from datetime import timedelta +from pathlib import Path from django import get_version as django_version from django.conf import settings @@ -103,6 +107,30 @@ def _fmt_bytes(n): size /= 1024 +def _folder_usage(path): + """Sum the allocated bytes of every file under ``path`` (like ``du -s``). + + Uses st_blocks (ext4/Unix allocation) when available, otherwise the + apparent file size. Returns None if the path doesn't exist or can't be + walked.""" + root = Path(path) + if not root.exists(): + return None + total = 0 + try: + for dirpath, dirnames, filenames in os.walk(root): + for name in filenames: + try: + st = os.stat(os.path.join(dirpath, name)) + except OSError: + continue + blocks = getattr(st, 'st_blocks', 0) + total += blocks * 512 if blocks else st.st_size + except OSError: + return None + return total + + def _system_snapshot(): """Collect a one-shot system/performance snapshot via psutil.""" try: @@ -137,12 +165,24 @@ def _system_snapshot(): for label, path in (('Root /', '/'), ('Code', str(settings.BASE_DIR)), ('Media', str(settings.MEDIA_ROOT))): try: d = psutil.disk_usage(path) - snapshot['disks'][label] = { - 'path': path, 'total': d.total, 'used': d.used, - 'free': d.free, 'percent': d.percent, - } except OSError: snapshot['disks'][label] = None + continue + # For folders (Code/Media) report the space the folder itself + # takes (like `du`), falling back to the whole-drive usage when the + # tree can't be measured. Root / always shows the drive. + used = d.used + scope = 'drive' + if label != 'Root /': + folder = _folder_usage(path) + if folder is not None: + used = folder + scope = 'folder' + snapshot['disks'][label] = { + 'path': path, 'total': d.total, 'used': used, + 'free': d.free, 'scope': scope, + 'percent': round(used / d.total * 100, 1) if d.total else 0, + } try: snapshot['net'] = { @@ -197,6 +237,97 @@ def server_stats(request): STATS_CACHE_KEY = 'stats:payload' STATS_CACHE_TTL = 20 +GUNICORN_ACCESS_LOG = Path(settings.BASE_DIR).parent / 'logs' / 'gunicorn-access.log' + +_ACCESS_LOG_LINE = re.compile( + r'^(?P\S+) \[(?P