mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-03 18:46:47 +09:00
Juan Denis
d7234b822d
Each was referenced only by its own definition across the panel, its tests, scripts, builtin extensions and every sibling extension repo (~350 lines). Imports they alone used go with them.
286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""Telemetry service for the unified system event stream.
|
|
|
|
This service provides a single, fire-and-forget way to record events from any
|
|
subsystem. It never raises into callers and redacts known sensitive keys from
|
|
payloads before storage.
|
|
"""
|
|
import json
|
|
import logging
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
|
|
from flask import g, has_request_context, request
|
|
from sqlalchemy import func
|
|
|
|
from app import db
|
|
from app.models.system_event import SystemEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from app.utils.sensitive_data_filter import ( # noqa: F401 (re-export)
|
|
REDACTED,
|
|
SENSITIVE_KEY_PARTS,
|
|
is_sensitive_key as _shared_is_sensitive_key,
|
|
)
|
|
|
|
MAX_STRING_LENGTH = 2000
|
|
MAX_LIST_ITEMS = 50
|
|
MAX_DICT_ITEMS = 50
|
|
MAX_PAYLOAD_BYTES = 32 * 1024
|
|
|
|
|
|
def generate_correlation_id():
|
|
"""Generate a unique correlation ID for grouping related events."""
|
|
return secrets.token_hex(16)
|
|
|
|
|
|
class TelemetryService:
|
|
"""Service for emitting and querying system telemetry events."""
|
|
|
|
# ------------------------------------------------------------------
|
|
# Emit
|
|
# ------------------------------------------------------------------
|
|
@classmethod
|
|
def emit(cls, source, event_type, message=None, severity=SystemEvent.SEVERITY_INFO,
|
|
resource_type=None, resource_id=None, actor_user_id=None, workspace_id=None,
|
|
correlation_id=None, payload=None, commit=True):
|
|
"""Record a telemetry event.
|
|
|
|
This method is fire-and-forget: any failure is logged but never raised,
|
|
so telemetry cannot break the calling operation.
|
|
"""
|
|
try:
|
|
event = cls._create_event(
|
|
source=source,
|
|
event_type=event_type,
|
|
message=message,
|
|
severity=severity,
|
|
resource_type=resource_type,
|
|
resource_id=_stringify_resource_id(resource_id),
|
|
actor_user_id=actor_user_id,
|
|
workspace_id=workspace_id,
|
|
correlation_id=correlation_id,
|
|
payload=payload,
|
|
)
|
|
db.session.add(event)
|
|
if commit:
|
|
db.session.commit()
|
|
return event
|
|
except Exception as exc:
|
|
logger.warning('Telemetry emit failed for %s.%s: %s', source, event_type, exc)
|
|
db.session.rollback()
|
|
return None
|
|
|
|
@classmethod
|
|
def _create_event(cls, source, event_type, message, severity, resource_type,
|
|
resource_id, actor_user_id, workspace_id, correlation_id, payload):
|
|
if severity not in SystemEvent.VALID_SEVERITIES:
|
|
severity = SystemEvent.SEVERITY_INFO
|
|
|
|
sanitized_payload = _sanitize_payload(payload) if payload is not None else None
|
|
if sanitized_payload is not None and _payload_too_large(sanitized_payload):
|
|
sanitized_payload = {'_truncated': True}
|
|
|
|
# Truncate very long messages
|
|
if message is not None and len(message) > 4000:
|
|
message = message[:4000] + '...[truncated]'
|
|
|
|
event = SystemEvent(
|
|
timestamp=datetime.utcnow(),
|
|
source=source,
|
|
event_type=event_type,
|
|
severity=severity,
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
actor_user_id=actor_user_id,
|
|
workspace_id=workspace_id,
|
|
message=message,
|
|
correlation_id=correlation_id,
|
|
)
|
|
event.set_payload(sanitized_payload)
|
|
return event
|
|
|
|
@staticmethod
|
|
def _get_request_scope():
|
|
"""Return (actor_user_id, workspace_id) from request context if available."""
|
|
if not has_request_context():
|
|
return None, None
|
|
|
|
actor_user_id = None
|
|
workspace_id = None
|
|
|
|
# JWT or API key user
|
|
try:
|
|
from flask_jwt_extended import get_jwt_identity
|
|
identity = get_jwt_identity()
|
|
if identity is not None:
|
|
try:
|
|
actor_user_id = int(identity)
|
|
except (TypeError, ValueError):
|
|
actor_user_id = identity
|
|
except Exception:
|
|
pass
|
|
|
|
if actor_user_id is None:
|
|
api_key_user = getattr(g, 'api_key_user', None)
|
|
if api_key_user is not None:
|
|
actor_user_id = api_key_user.id
|
|
|
|
# Workspace from g if set by other middleware
|
|
workspace_id = getattr(g, 'workspace_id', None)
|
|
|
|
return actor_user_id, workspace_id
|
|
|
|
# ------------------------------------------------------------------
|
|
# Query
|
|
# ------------------------------------------------------------------
|
|
@classmethod
|
|
def get_events(cls, page=1, per_page=50, source=None, event_type=None, severity=None,
|
|
resource_type=None, resource_id=None, correlation_id=None,
|
|
start_date=None, end_date=None, message_query=None):
|
|
"""Query system events with filters and pagination."""
|
|
query = SystemEvent.query.order_by(SystemEvent.timestamp.desc())
|
|
|
|
if source:
|
|
query = query.filter(SystemEvent.source == source)
|
|
if event_type:
|
|
query = query.filter(SystemEvent.event_type == event_type)
|
|
if severity:
|
|
query = query.filter(SystemEvent.severity == severity)
|
|
if resource_type:
|
|
query = query.filter(SystemEvent.resource_type == resource_type)
|
|
if resource_id is not None:
|
|
query = query.filter(SystemEvent.resource_id == _stringify_resource_id(resource_id))
|
|
if correlation_id:
|
|
query = query.filter(SystemEvent.correlation_id == correlation_id)
|
|
if start_date:
|
|
query = query.filter(SystemEvent.timestamp >= start_date)
|
|
if end_date:
|
|
query = query.filter(SystemEvent.timestamp <= end_date)
|
|
if message_query:
|
|
query = query.filter(SystemEvent.message.ilike(f'%{message_query}%'))
|
|
|
|
return query.paginate(page=page, per_page=per_page, error_out=False)
|
|
|
|
@classmethod
|
|
def get_event_by_id(cls, event_id):
|
|
"""Get a single event by ID."""
|
|
return db.session.get(SystemEvent, event_id)
|
|
|
|
@classmethod
|
|
def get_events_by_correlation(cls, correlation_id, limit=100):
|
|
"""Get all events sharing a correlation ID, oldest first."""
|
|
return (SystemEvent.query
|
|
.filter_by(correlation_id=correlation_id)
|
|
.order_by(SystemEvent.timestamp.asc())
|
|
.limit(limit)
|
|
.all())
|
|
|
|
# ------------------------------------------------------------------
|
|
# Stats
|
|
# ------------------------------------------------------------------
|
|
@classmethod
|
|
def get_stats(cls, hours=24, source=None):
|
|
"""Return aggregate stats for the given time window."""
|
|
cutoff = datetime.utcnow() - timedelta(hours=hours)
|
|
query = SystemEvent.query.filter(SystemEvent.timestamp >= cutoff)
|
|
if source:
|
|
query = query.filter(SystemEvent.source == source)
|
|
|
|
total = query.count()
|
|
|
|
by_severity = dict(
|
|
query.with_entities(SystemEvent.severity, func.count(SystemEvent.id))
|
|
.group_by(SystemEvent.severity)
|
|
.all()
|
|
)
|
|
|
|
by_source = dict(
|
|
query.with_entities(SystemEvent.source, func.count(SystemEvent.id))
|
|
.group_by(SystemEvent.source)
|
|
.all()
|
|
)
|
|
|
|
return {
|
|
'total': total,
|
|
'by_severity': by_severity,
|
|
'by_source': by_source,
|
|
'hours': hours,
|
|
}
|
|
|
|
@classmethod
|
|
def get_recent_event_types(cls, limit=20):
|
|
"""Return the most common event types in the last 24 hours."""
|
|
cutoff = datetime.utcnow() - timedelta(hours=24)
|
|
rows = (SystemEvent.query
|
|
.filter(SystemEvent.timestamp >= cutoff)
|
|
.with_entities(SystemEvent.event_type, func.count(SystemEvent.id))
|
|
.group_by(SystemEvent.event_type)
|
|
.order_by(func.count(SystemEvent.id).desc())
|
|
.limit(limit)
|
|
.all())
|
|
return [{'event_type': t, 'count': c} for t, c in rows]
|
|
|
|
# ------------------------------------------------------------------
|
|
# Cleanup
|
|
# ------------------------------------------------------------------
|
|
@classmethod
|
|
def cleanup_old_events(cls, days=90):
|
|
"""Delete events older than the specified number of days."""
|
|
cutoff = datetime.utcnow() - timedelta(days=days)
|
|
deleted = SystemEvent.query.filter(SystemEvent.timestamp < cutoff).delete()
|
|
db.session.commit()
|
|
return deleted
|
|
|
|
|
|
# ------------------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------------------
|
|
|
|
def _stringify_resource_id(resource_id):
|
|
"""Normalize resource IDs to strings to support int/UUID/mixed use."""
|
|
if resource_id is None:
|
|
return None
|
|
return str(resource_id)
|
|
|
|
|
|
def _sanitize_payload(value, key=None, depth=0):
|
|
"""Sanitize a payload dict/list, redacting sensitive keys and capping size."""
|
|
if key and _is_sensitive_key(key):
|
|
return REDACTED
|
|
if depth >= 4:
|
|
return '[truncated]'
|
|
|
|
if isinstance(value, dict):
|
|
result = {}
|
|
for item_key, item_value in list(value.items())[:MAX_DICT_ITEMS]:
|
|
safe_key = str(item_key)[:MAX_STRING_LENGTH]
|
|
result[safe_key] = _sanitize_payload(item_value, key=safe_key, depth=depth + 1)
|
|
return result
|
|
|
|
if isinstance(value, (list, tuple)):
|
|
result = [_sanitize_payload(item, depth=depth + 1) for item in value[:MAX_LIST_ITEMS]]
|
|
return result
|
|
|
|
if isinstance(value, str):
|
|
if len(value) > MAX_STRING_LENGTH:
|
|
return value[:MAX_STRING_LENGTH] + '...[truncated]'
|
|
return value
|
|
|
|
if value is None or isinstance(value, (bool, int, float)):
|
|
return value
|
|
|
|
return str(value)[:MAX_STRING_LENGTH]
|
|
|
|
|
|
def _is_sensitive_key(key):
|
|
return _shared_is_sensitive_key(str(key))
|
|
|
|
|
|
def _payload_too_large(payload):
|
|
"""Check whether a payload would exceed the storage limit when serialized."""
|
|
try:
|
|
return len(json.dumps(payload).encode('utf-8')) > MAX_PAYLOAD_BYTES
|
|
except (TypeError, ValueError):
|
|
return False
|