test2/Amnezia-Web-Panel-main
Template
1
2
Fork 1
Code Issues Pull Requests Packages Projects Releases 18 Wiki Activity

Speed up API reads with data cache and faster server check/stats.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
orohi
2026-07-26 06:07:44 +03:00
co-authored by Cursor
parent 2b28892a9b
commit fa3569f81f
3 changed files with 280 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+243 -188
View File
@@ -2,6 +2,7 @@ import os
import sys
import json
import logging
from collections import Counter
from dotenv import load_dotenv
load_dotenv()
@@ -167,6 +168,11 @@ load_translations()
DATA_LOCK = asyncio.Lock()
async def load_data_async():
"""Load panel state off the event loop (uses in-process cache)."""
return await asyncio.to_thread(load_data)
async def save_data_async(data):
"""Saves panel state to PostgreSQL in a thread-safe way."""
async with DATA_LOCK:
@@ -183,6 +189,37 @@ def get_ssh(server):
)
def get_current_user(request: Request, data: Optional[dict] = None):
user_id = request.session.get('user_id')
if not user_id:
return None
snapshot = data if data is not None else load_data()
for u in snapshot.get('users', []):
if u['id'] == user_id:
return u
return None
def tpl(request, template, **kwargs):
data = load_data()
lang = request.cookies.get('lang', 'en')
ctx = {
'request': request,
'current_user': get_current_user(request, data),
'site_settings': data.get('settings', {}).get('appearance', {}),
'captcha_settings': data.get('settings', {}).get('captcha', {}),
'telegram_settings': data.get('settings', {}).get('telegram', {}),
'bot_running': tg_bot.is_running(),
'lang': lang,
'_': lambda text_id: _t(text_id, lang),
'translations_json': json.dumps(TRANSLATIONS.get(lang, TRANSLATIONS.get('en', {}))),
# Keep for legacy JS; prefer translations_json on new pages
'all_translations_json': json.dumps(TRANSLATIONS),
}
ctx.update(kwargs)
return templates.TemplateResponse(template, ctx)
def get_panel_local_url(request: Optional[Request] = None):
data = load_data()
ssl_conf = data.get('settings', {}).get('ssl', {})
@@ -1337,36 +1374,6 @@ async def sync_users_with_remnawave(data: dict):
return 0, f"Error: {str(e)}"
def get_current_user(request: Request):
user_id = request.session.get('user_id')
if not user_id:
return None
data = load_data()
for u in data.get('users', []):
if u['id'] == user_id:
return u
return None
def tpl(request, template, **kwargs):
data = load_data()
lang = request.cookies.get('lang', 'en')
ctx = {
'request': request,
'current_user': get_current_user(request),
'site_settings': data.get('settings', {}).get('appearance', {}),
'captcha_settings': data.get('settings', {}).get('captcha', {}),
'telegram_settings': data.get('settings', {}).get('telegram', {}),
'bot_running': tg_bot.is_running(),
'lang': lang,
'_': lambda text_id: _t(text_id, lang),
'translations_json': json.dumps(TRANSLATIONS.get(lang, TRANSLATIONS.get('en', {}))),
'all_translations_json': json.dumps(TRANSLATIONS)
}
ctx.update(kwargs)
return templates.TemplateResponse(template, ctx)
# ======================== Pydantic Models ========================
class LoginRequest(BaseModel):
@@ -1946,22 +1953,24 @@ async def server_detail(request: Request, server_id: int):
@app.get('/users', response_class=HTMLResponse, tags=["System Templates"])
async def users_page(request: Request):
user = get_current_user(request)
data = load_data()
user = get_current_user(request, data)
if not user:
return RedirectResponse(url='/login', status_code=302)
if user['role'] not in ('admin', 'support'):
return RedirectResponse(url='/my', status_code=302)
data = load_data()
users_list = data.get('users', [])
# Count connections per user
conns = data.get('user_connections', [])
conn_counts = Counter(c.get('user_id') for c in conns if c.get('user_id'))
for u in users_list:
u['connections_count'] = sum(1 for c in conns if c['user_id'] == u['id'])
u['connections_count'] = conn_counts.get(u['id'], 0)
servers = data['servers']
return tpl(
request, 'users.html',
users=users_list,
servers=servers,
current_user=user,
)
@@ -2309,54 +2318,97 @@ rm -rf /opt/amnezia
return JSONResponse({'error': str(e)}, status_code=500)
def _docker_container_inventory(ssh) -> dict:
"""One SSH round-trip: map container name -> running bool."""
out, _, _ = ssh.run_sudo_command(
"docker ps -a --format '{{.Names}}|{{.Status}}' 2>/dev/null"
)
inventory = {}
for line in (out or '').splitlines():
if '|' not in line:
continue
name, status = line.split('|', 1)
name = name.strip()
if not name:
continue
inventory[name] = status.strip().lower().startswith('up')
return inventory
def _docker_is_ready(ssh) -> bool:
ver, _, vcode = ssh.run_command("docker --version 2>/dev/null")
return vcode == 0 and bool((ver or '').strip())
@app.post('/api/servers/{server_id}/stats', tags=["Servers"])
async def api_server_stats(request: Request, server_id: int):
if not _check_admin(request):
return JSONResponse({'error': 'Forbidden'}, status_code=403)
try:
data = load_data()
data = await load_data_async()
if server_id >= len(data['servers']):
return JSONResponse({'error': 'Server not found'}, status_code=404)
server = data['servers'][server_id]
ssh = get_ssh(server)
ssh.connect()
stats = {}
out, _, _ = ssh.run_command(
"top -bn1 | grep 'Cpu(s)' | awk '{print $2}' | cut -d'%' -f1 2>/dev/null || "
"awk '{u=$2+$4; t=$2+$4+$5; if(NR==1){pu=u;pt=t} else printf \"%.1f\", (u-pu)/(t-pt)*100}' "
"<(grep 'cpu ' /proc/stat) <(sleep 0.5 && grep 'cpu ' /proc/stat) 2>/dev/null"
)
try:
stats['cpu'] = round(float(out.strip().split('\n')[0]), 1)
except (ValueError, IndexError):
stats['cpu'] = 0
out, _, _ = ssh.run_command("free -b | awk 'NR==2{printf \"%d %d\", $3, $2}'")
try:
parts = out.strip().split()
used, total = int(parts[0]), int(parts[1])
stats.update(ram_used=used, ram_total=total, ram_percent=round(used / total * 100, 1) if total > 0 else 0)
except (ValueError, IndexError):
stats.update(ram_used=0, ram_total=0, ram_percent=0)
out, _, _ = ssh.run_command("df -B1 / | awk 'NR==2{printf \"%d %d\", $3, $2}'")
try:
parts = out.strip().split()
used, total = int(parts[0]), int(parts[1])
stats.update(disk_used=used, disk_total=total, disk_percent=round(used / total * 100, 1) if total > 0 else 0)
except (ValueError, IndexError):
stats.update(disk_used=0, disk_total=0, disk_percent=0)
out, _, _ = ssh.run_command(
"DEV=$(ip route | awk '/default/ {print $5}' | head -1); "
"cat /proc/net/dev | awk -v dev=\"$DEV:\" '$1==dev{printf \"%d %d\", $2, $10}'"
)
try:
parts = out.strip().split()
stats['net_rx'], stats['net_tx'] = int(parts[0]), int(parts[1])
except (ValueError, IndexError):
stats['net_rx'] = stats['net_tx'] = 0
out, _, _ = ssh.run_command("uptime -p 2>/dev/null || uptime")
stats['uptime'] = out.strip()
ssh.disconnect()
return stats
def _fetch_stats():
ssh = get_ssh(server)
ssh.connect()
try:
script = r"""
cpu=$(top -bn1 2>/dev/null | grep 'Cpu(s)' | awk '{print $2}' | cut -d'%' -f1)
if [ -z "$cpu" ]; then cpu=0; fi
ram=$(free -b 2>/dev/null | awk 'NR==2{printf "%d %d", $3, $2}')
disk=$(df -B1 / 2>/dev/null | awk 'NR==2{printf "%d %d", $3, $2}')
dev=$(ip route 2>/dev/null | awk '/default/ {print $5; exit}')
net=$(awk -v d="${dev}:" '$1==d{printf "%d %d", $2, $10}' /proc/net/dev 2>/dev/null)
up=$(uptime -p 2>/dev/null || uptime)
printf 'CPU|%s\nRAM|%s\nDISK|%s\nNET|%s\nUP|%s\n' "${cpu:-0}" "${ram:-0 0}" "${disk:-0 0}" "${net:-0 0}" "$up"
"""
out, _, _ = ssh.run_command(script)
finally:
ssh.disconnect()
stats = {
'cpu': 0,
'ram_used': 0, 'ram_total': 0, 'ram_percent': 0,
'disk_used': 0, 'disk_total': 0, 'disk_percent': 0,
'net_rx': 0, 'net_tx': 0,
'uptime': '',
}
for line in (out or '').splitlines():
if '|' not in line:
continue
key, val = line.split('|', 1)
key, val = key.strip(), val.strip()
try:
if key == 'CPU':
stats['cpu'] = round(float(val.split()[0] if val else 0), 1)
elif key == 'RAM':
parts = val.split()
used, total = int(parts[0]), int(parts[1])
stats.update(
ram_used=used,
ram_total=total,
ram_percent=round(used / total * 100, 1) if total > 0 else 0,
)
elif key == 'DISK':
parts = val.split()
used, total = int(parts[0]), int(parts[1])
stats.update(
disk_used=used,
disk_total=total,
disk_percent=round(used / total * 100, 1) if total > 0 else 0,
)
elif key == 'NET':
parts = val.split()
stats['net_rx'], stats['net_tx'] = int(parts[0]), int(parts[1])
elif key == 'UP':
stats['uptime'] = val
except (ValueError, IndexError):
continue
return stats
return await asyncio.to_thread(_fetch_stats)
except Exception as e:
logger.exception("Error getting server stats")
return JSONResponse({'error': str(e)}, status_code=500)
@@ -2367,128 +2419,130 @@ async def api_check_server(request: Request, server_id: int):
if not _check_admin(request):
return JSONResponse({'error': 'Forbidden'}, status_code=403)
try:
data = load_data()
data = await load_data_async()
if server_id >= len(data['servers']):
return JSONResponse({'error': 'Server not found'}, status_code=404)
server = data['servers'][server_id]
ssh = get_ssh(server)
ssh.connect()
# Just use awg's docker checker since it uses the same command
manager = get_protocol_manager(ssh, 'awg')
status = {'connection': 'ok', 'docker_installed': manager.check_docker_installed(), 'protocols': {}}
changed = False
if 'protocols' not in server:
server['protocols'] = {}
def merge_saved_protocol_status(proto, result=None, error=None):
"""Merge live status with saved protocol metadata.
Multi-instance protocols are source-of-truth in data.json because
they cannot be discovered from BASE_PROTOCOLS alone. A transient
check failure must not delete awg2__2/awg2__3 from the panel.
"""
db_proto = server.get('protocols', {}).get(proto, {}) or {}
merged = dict(result or {})
merged.setdefault('protocol', proto)
if error:
merged['error'] = error
if not merged.get('port') and db_proto.get('port'):
merged['port'] = db_proto['port']
if db_proto.get('awg_params') and not merged.get('awg_params'):
merged['awg_params'] = db_proto.get('awg_params')
merged['base_protocol'] = db_proto.get('base_protocol') or protocol_base(proto)
merged['instance'] = db_proto.get('instance') or protocol_instance(proto)
merged['display_name'] = db_proto.get('display_name') or protocol_display_name(proto)
merged['container_name'] = db_proto.get('container_name') or protocol_container_name(proto)
if protocol_base(proto) == 'adguard':
for key in ('web_port', 'mode', 'internal_ip', 'expose_web'):
if db_proto.get(key) not in (None, ''):
if key == 'web_port' and merged.get('web_exposed'):
continue
merged[key] = db_proto[key]
if 'expose_web' in db_proto and 'web_exposed' not in merged:
merged['web_exposed'] = bool(db_proto.get('expose_web'))
if protocol_base(proto) == 'nginx':
for key in ('domain', 'email', 'site_url'):
if db_proto.get(key) not in (None, ''):
merged[key] = db_proto[key]
return merged
def should_preserve_saved_protocol(proto, result=None, err=None):
"""Return True when check must not remove a saved protocol record."""
db_proto = server.get('protocols', {}).get(proto)
if not db_proto:
return False
# Additional AWG-family instances are only known by their saved
# dynamic keys (awg__2/awg2__2/awg_legacy__2). Keep them unless
# the user explicitly uninstalls them.
if protocol_base(proto) in MULTI_INSTANCE_PROTOCOLS and protocol_instance(proto) > 1:
return True
# Do not delete any saved protocol on command/check errors; only a
# clean live result with container_exists=False may prune base apps.
if err or (result and result.get('error')):
return True
return False
def check_proto(proto):
def _run_check():
ssh = get_ssh(server)
ssh.connect()
try:
p_manager = get_protocol_manager(ssh, proto)
result = _manager_call(p_manager, 'get_server_status', proto)
return proto, merge_saved_protocol_status(proto, result), None
except Exception as e:
return proto, merge_saved_protocol_status(proto, {}, str(e)), str(e)
status = {
'connection': 'ok',
'docker_installed': _docker_is_ready(ssh),
'protocols': {},
}
inventory = _docker_container_inventory(ssh) if status['docker_installed'] else {}
protocols_to_check = list(dict.fromkeys(BASE_PROTOCOLS + list(server.get('protocols', {}).keys())))
# Run checks sequentially. Several managers use the same SSH connection;
# checking them in parallel through one SSH object can produce false
# negatives and previously caused dynamic AWG instances to be removed.
for proto in protocols_to_check:
proto, result, err = check_proto(proto)
status['protocols'][proto] = result
if err:
continue
if result.get('container_exists'):
if proto not in server['protocols']:
server['protocols'][proto] = {
'installed': True,
'port': result.get('port', '55424'),
'awg_params': result.get('awg_params', {}),
'base_protocol': protocol_base(proto),
'instance': protocol_instance(proto),
'display_name': protocol_display_name(proto),
'container_name': protocol_container_name(proto),
}
changed = False
if 'protocols' not in server:
server['protocols'] = {}
def merge_saved_protocol_status(proto, result=None, error=None):
db_proto = server.get('protocols', {}).get(proto, {}) or {}
merged = dict(result or {})
merged.setdefault('protocol', proto)
if error:
merged['error'] = error
if not merged.get('port') and db_proto.get('port'):
merged['port'] = db_proto['port']
if db_proto.get('awg_params') and not merged.get('awg_params'):
merged['awg_params'] = db_proto.get('awg_params')
merged['base_protocol'] = db_proto.get('base_protocol') or protocol_base(proto)
merged['instance'] = db_proto.get('instance') or protocol_instance(proto)
merged['display_name'] = db_proto.get('display_name') or protocol_display_name(proto)
merged['container_name'] = db_proto.get('container_name') or protocol_container_name(proto)
if protocol_base(proto) == 'adguard':
server['protocols'][proto].update({
'mode': result.get('mode'),
'internal_ip': result.get('internal_ip'),
'web_port': result.get('web_port'),
'expose_web': result.get('web_exposed'),
})
for key in ('web_port', 'mode', 'internal_ip', 'expose_web'):
if db_proto.get(key) not in (None, ''):
if key == 'web_port' and merged.get('web_exposed'):
continue
merged[key] = db_proto[key]
if 'expose_web' in db_proto and 'web_exposed' not in merged:
merged['web_exposed'] = bool(db_proto.get('expose_web'))
if protocol_base(proto) == 'nginx':
server['protocols'][proto].update({
'domain': result.get('domain'),
'email': result.get('email'),
'site_url': result.get('site_url'),
})
changed = True
else:
if proto in server['protocols']:
if should_preserve_saved_protocol(proto, result, err):
# Keep saved dynamic instances visible as installed but stopped/unchecked.
status['protocols'][proto]['container_exists'] = True
status['protocols'][proto].setdefault('container_running', False)
status['protocols'][proto]['status_preserved'] = True
for key in ('domain', 'email', 'site_url'):
if db_proto.get(key) not in (None, ''):
merged[key] = db_proto[key]
return merged
def should_preserve_saved_protocol(proto, result=None, err=None):
db_proto = server.get('protocols', {}).get(proto)
if not db_proto:
return False
if protocol_base(proto) in MULTI_INSTANCE_PROTOCOLS and protocol_instance(proto) > 1:
return True
if err or (result and result.get('error')):
return True
return False
def check_proto(proto):
cname = protocol_container_name(proto)
# Fast path: skip deep manager probes when container is absent
if cname and cname not in inventory:
return proto, merge_saved_protocol_status(proto, {
'container_exists': False,
'container_running': False,
'protocol': proto,
}), None
try:
p_manager = get_protocol_manager(ssh, proto)
result = _manager_call(p_manager, 'get_server_status', proto)
return proto, merge_saved_protocol_status(proto, result), None
except Exception as e:
return proto, merge_saved_protocol_status(proto, {}, str(e)), str(e)
protocols_to_check = list(dict.fromkeys(
BASE_PROTOCOLS + list(server.get('protocols', {}).keys())
))
for proto in protocols_to_check:
proto, result, err = check_proto(proto)
status['protocols'][proto] = result
if err:
continue
if result.get('container_exists'):
if proto not in server['protocols']:
server['protocols'][proto] = {
'installed': True,
'port': result.get('port', '55424'),
'awg_params': result.get('awg_params', {}),
'base_protocol': protocol_base(proto),
'instance': protocol_instance(proto),
'display_name': protocol_display_name(proto),
'container_name': protocol_container_name(proto),
}
if protocol_base(proto) == 'adguard':
server['protocols'][proto].update({
'mode': result.get('mode'),
'internal_ip': result.get('internal_ip'),
'web_port': result.get('web_port'),
'expose_web': result.get('web_exposed'),
})
if protocol_base(proto) == 'nginx':
server['protocols'][proto].update({
'domain': result.get('domain'),
'email': result.get('email'),
'site_url': result.get('site_url'),
})
changed = True
else:
del server['protocols'][proto]
changed = True
if changed:
save_data(data)
ssh.disconnect()
return status
if proto in server['protocols']:
if should_preserve_saved_protocol(proto, result, err):
status['protocols'][proto]['container_exists'] = True
status['protocols'][proto].setdefault('container_running', False)
status['protocols'][proto]['status_preserved'] = True
else:
del server['protocols'][proto]
changed = True
if changed:
save_data(data)
return status
finally:
ssh.disconnect()
return await asyncio.to_thread(_run_check)
except Exception as e:
logger.exception("Error checking server")
return JSONResponse({'error': str(e), 'connection': 'failed'}, status_code=500)
@@ -3470,9 +3524,10 @@ async def api_toggle_connection(request: Request, server_id: int, req: ToggleCon
async def api_list_users(request: Request, search: str = '', page: int = 1, size: int = 10):
if not _check_admin(request):
return JSONResponse({'error': 'Forbidden'}, status_code=403)
data = load_data()
data = await load_data_async()
all_users = data.get('users', [])
conns = data.get('user_connections', [])
conn_counts = Counter(c.get('user_id') for c in conns if c.get('user_id'))
# Filter
filtered = []
@@ -3500,7 +3555,7 @@ async def api_list_users(request: Request, search: str = '', page: int = 1, size
'telegramId': u.get('telegramId'),
'email': u.get('email'),
'description': u.get('description'),
'connections_count': sum(1 for c in conns if c['user_id'] == u['id']),
'connections_count': conn_counts.get(u['id'], 0),
'traffic_used': u.get('traffic_used', 0),
'traffic_total': u.get('traffic_total', 0),
'traffic_limit': u.get('traffic_limit', 0),
db/store.py
+36 -1
View File
@@ -6,10 +6,12 @@ FastAPI handlers keep working without a full rewrite.
from __future__ import annotations
import copy
import json
import logging
import os
import shutil
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
@@ -21,6 +23,11 @@ from .connection import get_pool, init_schema
logger = logging.getLogger(__name__)
# Process-wide snapshot cache. load_data() returns a deep copy so callers can
# mutate safely; save_data() refreshes the cache after a successful write.
_DATA_CACHE: Optional[dict] = None
_DATA_CACHE_LOCK = threading.RLock()
DEFAULT_SETTINGS = {
'appearance': {
'title': 'Amnezia',
@@ -208,7 +215,13 @@ def _row_to_invite(row) -> dict:
}
def load_data() -> dict:
def invalidate_data_cache() -> None:
global _DATA_CACHE
with _DATA_CACHE_LOCK:
_DATA_CACHE = None
def _fetch_data_from_db() -> dict:
init_schema()
pool = get_pool()
with pool.connection() as conn:
@@ -266,8 +279,20 @@ def load_data() -> dict:
}
def load_data() -> dict:
"""Return panel state. Uses an in-process cache; always returns a deep copy."""
global _DATA_CACHE
with _DATA_CACHE_LOCK:
if _DATA_CACHE is not None:
return copy.deepcopy(_DATA_CACHE)
data = _fetch_data_from_db()
_DATA_CACHE = data
return copy.deepcopy(data)
def save_data(data: dict) -> None:
"""Replace panel state in a single transaction (same semantics as rewriting data.json)."""
global _DATA_CACHE
init_schema()
servers = data.get('servers') or []
users = data.get('users') or []
@@ -419,6 +444,16 @@ def save_data(data: dict) -> None:
)
conn.commit()
# Keep caller dict and cache aligned with what was actually persisted
data['servers'] = servers
data['users'] = users
data['user_connections'] = connections
data['api_tokens'] = tokens
data['invite_links'] = invite_links
data['settings'] = settings
with _DATA_CACHE_LOCK:
_DATA_CACHE = copy.deepcopy(data)
def export_data_dict() -> dict:
"""Export current DB state as the legacy data.json document."""
managers/ssh_manager.py
+1 -1
View File
@@ -66,7 +66,7 @@ class SSHManager:
if not self.client:
raise ConnectionError("Not connected to server")
logger.info(f"Running command: {command[:100]}...")
logger.debug(f"Running command: {command[:100]}...")
stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout)
# Crucial: set timeout on the channel to prevent hanging indefinitely