Cheng Zhou
83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
"""权限指标小时聚合任务
|
|
|
|
每小时从 permission_access_logs 聚合数据写入 permission_metric_snapshots。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.models.permission_access_log import PermissionAccessLog
|
|
from app.models.permission_metric_snapshot import PermissionMetricSnapshot
|
|
|
|
logger = logging.getLogger("ctms.permission_metric_aggregator")
|
|
|
|
|
|
async def _aggregate_hour(bucket_start: datetime, bucket_end: datetime) -> None:
|
|
async with SessionLocal() as session:
|
|
result = await session.execute(
|
|
select(
|
|
func.count().label("total"),
|
|
func.count().filter(PermissionAccessLog.allowed.is_(True)).label("allowed"),
|
|
func.count().filter(PermissionAccessLog.allowed.is_(False)).label("denied"),
|
|
func.coalesce(func.avg(PermissionAccessLog.elapsed_ms), 0).label("avg_ms"),
|
|
func.coalesce(func.max(PermissionAccessLog.elapsed_ms), 0).label("max_ms"),
|
|
).where(
|
|
PermissionAccessLog.created_at >= bucket_start,
|
|
PermissionAccessLog.created_at < bucket_end,
|
|
)
|
|
)
|
|
row = result.one()
|
|
|
|
if row.total == 0:
|
|
return
|
|
|
|
from app.core.permission_monitor import get_permission_monitor
|
|
monitor = get_permission_monitor()
|
|
cache_metrics = monitor.metrics.cache_metrics.window_stats(3600)
|
|
|
|
snapshot = PermissionMetricSnapshot(
|
|
id=uuid.uuid4(),
|
|
bucket_time=bucket_start,
|
|
total_checks=row.total,
|
|
allowed_checks=row.allowed,
|
|
denied_checks=row.denied,
|
|
avg_elapsed_ms=float(row.avg_ms),
|
|
max_elapsed_ms=float(row.max_ms),
|
|
cache_hits=cache_metrics["cache_hits"],
|
|
cache_misses=cache_metrics["cache_misses"],
|
|
error_count=0,
|
|
)
|
|
session.add(snapshot)
|
|
await session.commit()
|
|
logger.info("Aggregated permission metrics for bucket %s: %d checks", bucket_start, row.total)
|
|
|
|
|
|
async def run_hourly_metric_aggregation(stop_event: asyncio.Event) -> None:
|
|
logger.info("Permission metric aggregator started")
|
|
while not stop_event.is_set():
|
|
now = datetime.now(timezone.utc)
|
|
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
|
|
wait_seconds = (next_hour - now).total_seconds()
|
|
|
|
try:
|
|
await asyncio.wait_for(stop_event.wait(), timeout=wait_seconds)
|
|
break
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
bucket_end = next_hour
|
|
bucket_start = bucket_end - timedelta(hours=1)
|
|
try:
|
|
await _aggregate_hour(bucket_start, bucket_end)
|
|
except Exception:
|
|
logger.exception("Failed to aggregate permission metrics for %s", bucket_start)
|
|
|
|
logger.info("Permission metric aggregator stopped")
|