97 lines
2.4 KiB
Python
97 lines
2.4 KiB
Python
import json
|
|
import os
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
class LogEntry(BaseModel):
|
|
ts: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
|
level: str = "INFO"
|
|
service: str
|
|
event: str
|
|
message: Optional[str] = None
|
|
trace_id: Optional[str] = None
|
|
team_id: Optional[int] = None
|
|
user_id: Optional[int] = None
|
|
match_id: Optional[int] = None
|
|
round_id: Optional[int] = None
|
|
service_id: Optional[str] = None
|
|
status: Optional[str] = None
|
|
extra: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
@field_validator("level")
|
|
@classmethod
|
|
def validate_level(cls, value: str) -> str:
|
|
value = value.upper()
|
|
allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
|
|
|
|
if value not in allowed:
|
|
raise ValueError(
|
|
"level must be DEBUG, INFO, WARNING, ERROR or CRITICAL"
|
|
)
|
|
|
|
return value
|
|
|
|
|
|
class JsonlStorage:
|
|
def __init__(self, path: str):
|
|
self.path = Path(path)
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.Lock()
|
|
|
|
def write(self, record: Dict[str, Any]) -> None:
|
|
line = json.dumps(
|
|
record,
|
|
ensure_ascii=False,
|
|
default=str,
|
|
)
|
|
|
|
with self._lock:
|
|
with self.path.open("a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
|
|
|
|
storage = JsonlStorage(
|
|
os.getenv("LOG_STORAGE_PATH", "./data/logs.jsonl")
|
|
)
|
|
|
|
app = FastAPI(
|
|
title="CTF Log Collector",
|
|
version="0.1.0",
|
|
)
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> Dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/logs", status_code=202)
|
|
def collect_log(entry: LogEntry) -> Dict[str, str]:
|
|
payload = entry.model_dump(mode="json")
|
|
storage.write(payload)
|
|
|
|
return {"status": "accepted"}
|
|
|
|
|
|
@app.post("/logs/batch", status_code=202)
|
|
def collect_logs_batch(entries: List[LogEntry]) -> Dict[str, Any]:
|
|
if len(entries) > 1000:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Too many log entries in one batch. Max is 1000.",
|
|
)
|
|
|
|
for entry in entries:
|
|
payload = entry.model_dump(mode="json")
|
|
storage.write(payload)
|
|
|
|
return {
|
|
"status": "accepted",
|
|
"count": len(entries),
|
|
} |