Cheng Zhou
b491b6a146
Client Quality Gates / Shared client and Web (push) Has been cancelled
Client Quality Gates / macOS Desktop (push) Has been cancelled
Storage Persistence Guard / storage-persistence-audit (push) Has been cancelled
Client Quality Gates / Shared client and Web (pull_request) Has been cancelled
Client Quality Gates / macOS Desktop (pull_request) Has been cancelled
Storage Persistence Guard / storage-persistence-audit (pull_request) Has been cancelled
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""底层安全访问日志后台写入器"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.models.security_access_log import SecurityAccessLog
|
|
|
|
logger = logging.getLogger("ctms.security_access_log_writer")
|
|
|
|
BATCH_SIZE = 100
|
|
FLUSH_INTERVAL = 3.0
|
|
QUEUE_MAX_SIZE = 20000
|
|
|
|
|
|
class SecurityAccessLogWriter:
|
|
def __init__(self) -> None:
|
|
self._queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
|
|
self._task: asyncio.Task | None = None
|
|
|
|
def enqueue(self, entry: dict) -> None:
|
|
try:
|
|
self._queue.put_nowait(entry)
|
|
except asyncio.QueueFull:
|
|
logger.warning("Security access log queue full, dropping entry")
|
|
|
|
async def start(self) -> None:
|
|
self._task = asyncio.create_task(self._flush_loop())
|
|
logger.info("SecurityAccessLogWriter started")
|
|
|
|
async def stop(self) -> None:
|
|
if self._task:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
await self._drain()
|
|
logger.info("SecurityAccessLogWriter stopped")
|
|
|
|
async def _flush_loop(self) -> None:
|
|
while True:
|
|
batch = await self._collect_batch()
|
|
if batch:
|
|
await self._write_batch(batch)
|
|
|
|
async def _collect_batch(self) -> list[dict]:
|
|
batch: list[dict] = []
|
|
try:
|
|
first = await asyncio.wait_for(self._queue.get(), timeout=FLUSH_INTERVAL)
|
|
batch.append(first)
|
|
except asyncio.TimeoutError:
|
|
return batch
|
|
|
|
while len(batch) < BATCH_SIZE:
|
|
try:
|
|
batch.append(self._queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
return batch
|
|
|
|
async def _write_batch(self, batch: list[dict]) -> None:
|
|
try:
|
|
async with SessionLocal() as session:
|
|
for entry in batch:
|
|
session.add(
|
|
SecurityAccessLog(
|
|
id=uuid.uuid4(),
|
|
method=entry["method"],
|
|
path=entry["path"],
|
|
status_code=entry["status_code"],
|
|
elapsed_ms=entry["elapsed_ms"],
|
|
client_ip=entry.get("client_ip"),
|
|
user_agent=entry.get("user_agent"),
|
|
client_type=entry.get("client_type"),
|
|
client_version=entry.get("client_version"),
|
|
client_platform=entry.get("client_platform"),
|
|
build_channel=entry.get("build_channel"),
|
|
build_commit=entry.get("build_commit"),
|
|
auth_status=entry["auth_status"],
|
|
user_identifier=entry.get("user_identifier"),
|
|
created_at=entry.get("created_at", datetime.now(timezone.utc)),
|
|
)
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
logger.exception("Failed to write security access log batch (%d entries)", len(batch))
|
|
|
|
async def _drain(self) -> None:
|
|
batch: list[dict] = []
|
|
while not self._queue.empty():
|
|
try:
|
|
batch.append(self._queue.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
if batch:
|
|
await self._write_batch(batch)
|
|
|
|
|
|
_writer: SecurityAccessLogWriter | None = None
|
|
|
|
|
|
def get_security_log_writer() -> SecurityAccessLogWriter | None:
|
|
return _writer
|
|
|
|
|
|
async def start_security_log_writer() -> SecurityAccessLogWriter:
|
|
global _writer
|
|
_writer = SecurityAccessLogWriter()
|
|
await _writer.start()
|
|
return _writer
|
|
|
|
|
|
async def stop_security_log_writer() -> None:
|
|
global _writer
|
|
if _writer:
|
|
await _writer.stop()
|
|
_writer = None
|