feat: 接入 LogHive 远程日志系统
- 添加 LogHiveHandler，启动时自动挂载到 root logger - 所有 logging 日志自动异步发送到 LogHive，不影响主业务 - vendored loghive-client 包，Docker 构建时自动安装 - API Key 缺失时自动跳过，不影响本地开发
This commit is contained in:
Vendored
+183
@@ -0,0 +1,183 @@
|
||||
"""Synchronous LogHive client — uses threading for non-blocking sends."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from datetime import datetime, timezone
|
||||
from queue import Queue, Empty
|
||||
from typing import Any, Dict, List, Optional
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LogHiveLogger:
|
||||
"""Synchronous logger that sends logs to LogHive in the background.
|
||||
|
||||
Uses a background thread with a queue to avoid blocking the main
|
||||
application on network I/O.
|
||||
|
||||
Usage:
|
||||
logger = LogHiveLogger("my-project", "api-key-here", "http://localhost:8000")
|
||||
logger.info("Hello, world!")
|
||||
logger.error("Something broke", exc_info=True)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
project: str,
|
||||
api_key: str,
|
||||
endpoint: str = "http://localhost:8000",
|
||||
batch_size: int = 50,
|
||||
flush_interval: float = 2.0,
|
||||
max_retries: int = 3,
|
||||
timeout: float = 5.0,
|
||||
auto_start: bool = True,
|
||||
):
|
||||
self.project = project
|
||||
self.api_key = api_key
|
||||
self.endpoint = endpoint.rstrip("/")
|
||||
self.batch_size = batch_size
|
||||
self.flush_interval = flush_interval
|
||||
self.max_retries = max_retries
|
||||
self.timeout = timeout
|
||||
|
||||
self._queue: Queue = Queue()
|
||||
self._stop_event = threading.Event()
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
|
||||
if auto_start:
|
||||
self.start()
|
||||
|
||||
def start(self):
|
||||
"""Start the background flush thread."""
|
||||
if self._thread and self._thread.is_alive():
|
||||
return
|
||||
self._stop_event.clear()
|
||||
self._thread = threading.Thread(target=self._flush_loop, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def stop(self, flush: bool = True):
|
||||
"""Stop the background thread, optionally flushing remaining logs."""
|
||||
self._stop_event.set()
|
||||
if flush and self._thread:
|
||||
self._flush_now()
|
||||
if self._thread:
|
||||
self._thread.join(timeout=5)
|
||||
|
||||
def _flush_loop(self):
|
||||
"""Background loop that periodically flushes the queue."""
|
||||
while not self._stop_event.is_set():
|
||||
self._flush_now()
|
||||
self._stop_event.wait(self.flush_interval)
|
||||
|
||||
def _flush_now(self):
|
||||
"""Flush all currently queued log entries."""
|
||||
entries = []
|
||||
while len(entries) < self.batch_size:
|
||||
try:
|
||||
entry = self._queue.get_nowait()
|
||||
entries.append(entry)
|
||||
except Empty:
|
||||
break
|
||||
|
||||
if not entries:
|
||||
return
|
||||
|
||||
self._send_batch(entries)
|
||||
|
||||
def _send_batch(self, entries: List[Dict[str, Any]]):
|
||||
"""Send a batch of entries to the LogHive API, with retries."""
|
||||
url = urljoin(self.endpoint, "/api/logs/ingest")
|
||||
payload = {"project": self.project, "entries": entries}
|
||||
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
resp = httpx.post(
|
||||
url,
|
||||
json=payload,
|
||||
headers={"Authorization": f"Bearer {self.api_key}"},
|
||||
timeout=self.timeout,
|
||||
)
|
||||
if resp.status_code == 201:
|
||||
return
|
||||
elif resp.status_code == 401:
|
||||
logger.error("LogHive: Invalid API key — dropping batch")
|
||||
return
|
||||
else:
|
||||
logger.warning(
|
||||
"LogHive: HTTP %d (attempt %d/%d)",
|
||||
resp.status_code,
|
||||
attempt + 1,
|
||||
self.max_retries,
|
||||
)
|
||||
except httpx.RequestError as e:
|
||||
logger.warning(
|
||||
"LogHive: Connection error (attempt %d/%d): %s",
|
||||
attempt + 1,
|
||||
self.max_retries,
|
||||
e,
|
||||
)
|
||||
|
||||
if attempt < self.max_retries - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
|
||||
logger.error("LogHive: Failed to send %d entries after %d retries", len(entries), self.max_retries)
|
||||
|
||||
def _enqueue(self, level: str, message: str, **kwargs):
|
||||
"""Enqueue a log entry for async sending."""
|
||||
entry = {
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"level": level,
|
||||
"message": message,
|
||||
"logger": kwargs.pop("logger", None) or __name__,
|
||||
"module": kwargs.pop("module", None),
|
||||
"function": kwargs.pop("function", None),
|
||||
"line_no": kwargs.pop("line_no", None),
|
||||
"trace_id": kwargs.pop("trace_id", None),
|
||||
"extra": kwargs,
|
||||
}
|
||||
|
||||
# Handle exception info
|
||||
exc_info = kwargs.pop("exc_info", None)
|
||||
if exc_info:
|
||||
if isinstance(exc_info, BaseException):
|
||||
entry["exception"] = "".join(
|
||||
traceback.format_exception(type(exc_info), exc_info, exc_info.__traceback__)
|
||||
)
|
||||
elif exc_info is True:
|
||||
entry["exception"] = traceback.format_exc()
|
||||
|
||||
self._queue.put_nowait(entry)
|
||||
|
||||
# ── Public API (matching standard logging levels) ──────────
|
||||
|
||||
def debug(self, message: str, **kwargs):
|
||||
self._enqueue("debug", message, **kwargs)
|
||||
|
||||
def info(self, message: str, **kwargs):
|
||||
self._enqueue("info", message, **kwargs)
|
||||
|
||||
def warning(self, message: str, **kwargs):
|
||||
self._enqueue("warning", message, **kwargs)
|
||||
|
||||
def error(self, message: str, **kwargs):
|
||||
self._enqueue("error", message, **kwargs)
|
||||
|
||||
def critical(self, message: str, **kwargs):
|
||||
self._enqueue("critical", message, **kwargs)
|
||||
|
||||
def log(self, level: str, message: str, **kwargs):
|
||||
"""Log a message with an explicit level string."""
|
||||
self._enqueue(level, message, **kwargs)
|
||||
|
||||
def set_trace_id(self, trace_id: str):
|
||||
"""Set a trace_id for request correlation (used in web frameworks)."""
|
||||
self._current_trace_id = trace_id
|
||||
|
||||
def __del__(self):
|
||||
self.stop(flush=True)
|
||||
Reference in New Issue
Block a user