From e355c12ffa936dcbdbe9eaf7a30a9a9fd29fa6e5 Mon Sep 17 00:00:00 2001 From: dv Date: Thu, 30 Apr 2026 15:54:19 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BF=D0=BE=D0=BF=D1=8B=D1=82=D0=BA=D0=B0=20?= =?UTF-8?q?=D1=84=D0=B8=D0=BA=D1=81=D0=B0=20worker=20overflow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 5 +++++ backend/app/config.py | 7 +++++++ backend/app/services/database.py | 5 ++++- backend/app/services/ssh.py | 30 ++++++++++++++++++++++++------ backend/app/services/task.py | 20 ++++++++++---------- backend/app/workers/monitor.py | 13 +++++++++++-- 6 files changed, 61 insertions(+), 19 deletions(-) diff --git a/.env.example b/.env.example index d2d4277..d96ef7a 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,10 @@ # ── Database ────────────────────────────────────────────────────────────────── DATABASE_URL=postgresql+asyncpg://fleet:fleet@localhost:5432/fleet +DATABASE_POOL_SIZE=10 +DATABASE_MAX_OVERFLOW=20 +DATABASE_POOL_TIMEOUT=30 +TASK_DEVICE_CONCURRENCY=10 +MONITOR_PING_CONCURRENCY=20 # ── Security ────────────────────────────────────────────────────────────────── # Generate with: python -c "import secrets; print(secrets.token_hex(32))" diff --git a/backend/app/config.py b/backend/app/config.py index 31ef29c..9f0629e 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -17,5 +17,12 @@ class Settings(BaseSettings): SQLALCHEMY_LOG: bool = False + DATABASE_POOL_SIZE: int = 10 + DATABASE_MAX_OVERFLOW: int = 20 + DATABASE_POOL_TIMEOUT: int = 30 + + TASK_DEVICE_CONCURRENCY: int = 10 + MONITOR_PING_CONCURRENCY: int = 20 + settings = Settings() diff --git a/backend/app/services/database.py b/backend/app/services/database.py index 07a83a9..0ce91d8 100644 --- a/backend/app/services/database.py +++ b/backend/app/services/database.py @@ -8,7 +8,10 @@ engine = create_async_engine( settings.DATABASE_URL, #echo=settings.DEBUG, echo=False, - pool_pre_ping=True + pool_pre_ping=True, + pool_size=settings.DATABASE_POOL_SIZE, + max_overflow=settings.DATABASE_MAX_OVERFLOW, + pool_timeout=settings.DATABASE_POOL_TIMEOUT, ) diff --git a/backend/app/services/ssh.py b/backend/app/services/ssh.py index a101bd8..c5c8c9c 100644 --- a/backend/app/services/ssh.py +++ b/backend/app/services/ssh.py @@ -29,6 +29,7 @@ from app.models.device import Device from app.models.object import Object from app.models.ssh_key import SSHKey from app.services.crypto import decrypt +from app.services.database import AsyncSessionLocal logger = logging.getLogger(__name__) @@ -355,7 +356,11 @@ async def run_command( async with _global_sem: async with _get_object_sem(device.object_id): async with _get_device_sem(device.id): - options = await build_auth_options(device, obj, db) + if db: + options = await build_auth_options(device, obj, db) + else: + async with AsyncSessionLocal() as auth_db: + options = await build_auth_options(device, obj, auth_db) if not options: # Fallback: no creds configured at all @@ -389,16 +394,29 @@ async def run_command( ) if db: await _save_last_auth(device, db, opt) + else: + async with AsyncSessionLocal() as save_db: + async with save_db.begin(): + current_device = await save_db.get(Device, device.id) + if current_device: + await _save_last_auth(current_device, save_db, opt) return result # Auth failed (wrong password / key / connection error) # If last_auth method failed, clear it - if i == 0 and opt is last_auth_option and db: + if i == 0 and opt is last_auth_option: device.ssh_last_auth = None - try: - await db.flush() - except Exception: - pass + if db: + try: + await db.flush() + except Exception: + pass + else: + async with AsyncSessionLocal() as clear_db: + async with clear_db.begin(): + current_device = await clear_db.get(Device, device.id) + if current_device: + current_device.ssh_last_auth = None last_result = result logger.info( diff --git a/backend/app/services/task.py b/backend/app/services/task.py index f005967..c01128d 100644 --- a/backend/app/services/task.py +++ b/backend/app/services/task.py @@ -20,11 +20,12 @@ from app.models.rack import Rack from app.models.task import Task, TaskEvent, TaskResult from app.models.user import User from app.plugins.registry import ActionRegistry, action_registry +from app.config import settings from app.services.database import AsyncSessionLocal from app.services.sse import SSEEvent, sse_manager # Max parallel devices within a single task -_TASK_DEVICE_CONCURRENCY = 20 +_TASK_DEVICE_CONCURRENCY = settings.TASK_DEVICE_CONCURRENCY async def _emit(task_id: str, message: str, level: str = "info") -> None: @@ -115,13 +116,14 @@ async def _run_on_device( params: dict, sem: asyncio.Semaphore, ) -> bool | None: - # Bail out before acquiring the semaphore if task was cancelled - async with AsyncSessionLocal() as db: - task_row = await db.get(Task, task_id) - if task_row and task_row.cancel_requested: - return None # None = skipped due to cancellation - async with sem: + # Bail out after acquiring the semaphore so large target lists don't + # stampede the DB pool with cancellation checks. + async with AsyncSessionLocal() as db: + task_row = await db.get(Task, task_id) + if task_row and task_row.cancel_requested: + return None # None = skipped due to cancellation + emit = lambda msg, level="info": _emit(task_id, msg, level) # Create task_result row @@ -140,9 +142,7 @@ async def _run_on_device( result_id = tr.id try: - async with AsyncSessionLocal() as action_db: - async with action_db.begin(): - action_result = await action.execute(device, obj, params, emit, db=action_db) + action_result = await action.execute(device, obj, params, emit, db=None) except Exception as exc: action_result_success = False action_stdout = "" diff --git a/backend/app/workers/monitor.py b/backend/app/workers/monitor.py index 9501448..e558fb0 100644 --- a/backend/app/workers/monitor.py +++ b/backend/app/workers/monitor.py @@ -21,13 +21,14 @@ from sqlalchemy import select, update from app.models.alert import Alert from app.models.device import Device +from app.config import settings from app.services.database import AsyncSessionLocal from app.services.sse import SSEEvent, sse_manager # ── Configuration ────────────────────────────────────────────────────────────── MONITOR_INTERVAL = 60 # seconds between full check cycles PING_TIMEOUT = 3 # seconds per ping attempt -PING_CONCURRENCY = 30 # max simultaneous pings +PING_CONCURRENCY = settings.MONITOR_PING_CONCURRENCY # max simultaneous device checks MONITOR_CHANNEL = "__monitor__" @@ -98,7 +99,15 @@ async def _check_device( sem: asyncio.Semaphore, ) -> None: async with sem: - success, ping_ms = await _ping(ip) + await _check_device_locked(device_id, ip, object_id) + + +async def _check_device_locked( + device_id: int, + ip: str, + object_id: int, +) -> None: + success, ping_ms = await _ping(ip) now = datetime.now(timezone.utc) new_status = "online" if success else "offline"