Initial commit

log-collector/app.py

@@ -0,0 +1,97 @@
+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),
+    }