diff --git a/.env.example b/.env.example index d0ecd88..3d1b1eb 100644 --- a/.env.example +++ b/.env.example @@ -25,6 +25,11 @@ WECHAT_HOST=0.0.0.0 SCHEDULER_ENABLED=true SCHEDULER_CRON=0 8-21 * * * +# LogHive 日志系统 +LOGHIVE_ENDPOINT=http://10.10.10.14:8000 +LOGHIVE_PROJECT=gx-gp-notify +LOGHIVE_API_KEY= + # Markdown MARKDOWN_ENABLED=true MARKDOWN_OUTPUT_FILE=onu.md diff --git a/app/config.py b/app/config.py index c4cef54..0ae1bea 100644 --- a/app/config.py +++ b/app/config.py @@ -33,6 +33,11 @@ class Settings(BaseSettings): scheduler_enabled: bool = True scheduler_cron: str = "0 8-21 * * *" + # LogHive 日志系统 + loghive_endpoint: str = "http://10.10.10.14:8000" + loghive_project: str = "gx-gp-notify" + loghive_api_key: str = "" + # Markdown markdown_enabled: bool = True markdown_output_file: str = "onu.md" diff --git a/app/main.py b/app/main.py index de27de2..13173e1 100644 --- a/app/main.py +++ b/app/main.py @@ -22,6 +22,17 @@ async def lifespan(app: FastAPI): level=getattr(logging, settings.log_level), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) + # LogHive 远程日志(仅当配置了 API Key 时启用) + if settings.loghive_api_key: + from loghive_client import LogHiveHandler + handler = LogHiveHandler( + project=settings.loghive_project, + api_key=settings.loghive_api_key, + endpoint=settings.loghive_endpoint, + level=logging.INFO, + ) + logging.getLogger().addHandler(handler) + logging.info("LogHive 日志系统已连接") from app.scheduler.jobs import shutdown_scheduler, start_scheduler start_scheduler() yield diff --git a/docker/Dockerfile b/docker/Dockerfile index 9191c99..c7d0475 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -7,7 +7,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ rm -rf /var/lib/apt/lists/* COPY pyproject.toml . -RUN pip install --no-cache-dir -e ".[dev]" +COPY vendor/ vendor/ +RUN pip install --no-cache-dir -e ".[dev]" && pip install --no-cache-dir ./vendor/ COPY . . diff --git a/vendor/README.md b/vendor/README.md new file mode 100644 index 0000000..6e774fc --- /dev/null +++ b/vendor/README.md @@ -0,0 +1,84 @@ +# LogHive Client SDK + +Python client SDK for sending logs to [LogHive](https://github.com/your-org/loghive). + +## Installation + +```bash +pip install loghive-client +``` + +Or install from source: + +```bash +cd client +pip install . +``` + +## Quick Start + +### Sync mode (recommended for scripts, Django, Flask) + +```python +from loghive_client import LogHiveLogger + +logger = LogHiveLogger( + project="my-awesome-app", + api_key="your-api-key", + endpoint="http://localhost:8000", +) + +logger.info("Server started", extra={"port": 8080}) +logger.error("Database timeout", exc_info=True) +``` + +### Async mode (for FastAPI, aiohttp, asyncio) + +```python +from loghive_client import AsyncLogHiveLogger +import asyncio + +async def main(): + async with AsyncLogHiveLogger( + project="my-api", + api_key="your-api-key", + endpoint="http://localhost:8000", + ) as logger: + await logger.info("API started") + # ... + +asyncio.run(main()) +``` + +### Standard logging integration (zero code change) + +Add the handler to your existing logger: + +```python +import logging +from loghive_client import LogHiveHandler + +handler = LogHiveHandler("my-project", "api-key", "http://localhost:8000") +logging.getLogger().addHandler(handler) + +# All existing logger calls now forward to LogHive +logging.info("This goes to LogHive too!") +``` + +## Configuration + +| Param | Default | Description | +|-------|---------|-------------| +| `project` | (required) | Your project name in LogHive | +| `api_key` | (required) | Your project's API key | +| `endpoint` | `http://localhost:8000` | LogHive server URL | +| `batch_size` | 50 | Max entries per HTTP request | +| `flush_interval` | 2.0 | Seconds between flushes | +| `max_retries` | 3 | Retries on failure | +| `timeout` | 5.0 | HTTP request timeout | + +## Trace ID (request correlation) + +```python +logger.set_trace_id("req-abc-123") +``` diff --git a/vendor/loghive_client/__init__.py b/vendor/loghive_client/__init__.py new file mode 100644 index 0000000..af3e424 --- /dev/null +++ b/vendor/loghive_client/__init__.py @@ -0,0 +1,20 @@ +"""LogHive Client — Send logs from your Python projects to LogHive. + +Usage: + from loghive_client import LogHiveLogger + + logger = LogHiveLogger( + project="my-project", + api_key="your-api-key", + endpoint="http://localhost:8000", + ) + + logger.info("User logged in", extra={"user_id": 42}) + logger.error("Database connection failed", exc_info=True) +""" + +from loghive_client.client import LogHiveLogger +from loghive_client.async_client import AsyncLogHiveLogger +from loghive_client.handler import LogHiveHandler + +__all__ = ["LogHiveLogger", "AsyncLogHiveLogger", "LogHiveHandler"] diff --git a/vendor/loghive_client/async_client.py b/vendor/loghive_client/async_client.py new file mode 100644 index 0000000..22475f8 --- /dev/null +++ b/vendor/loghive_client/async_client.py @@ -0,0 +1,178 @@ +"""Async LogHive client — for use in asyncio-based projects (e.g., FastAPI, aiohttp).""" + +import asyncio +import logging +import traceback +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + + +class AsyncLogHiveLogger: + """Async logger for asyncio applications. + + Uses an async background task to batch and send log entries. + Ideal for FastAPI / Starlette / aiohttp projects. + + Usage: + logger = AsyncLogHiveLogger("my-project", "api-key", "http://localhost:8000") + await logger.start() + + await logger.info("Request processed", extra={"path": "/api/users"}) + + await logger.stop() + """ + + 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, + ): + 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: asyncio.Queue = asyncio.Queue() + self._task: Optional[asyncio.Task] = None + self._client: Optional[httpx.AsyncClient] = None + self._stop_event = asyncio.Event() + + async def start(self): + """Start the background flush task.""" + if self._task and not self._task.done(): + return + self._client = httpx.AsyncClient(timeout=self.timeout) + self._stop_event.clear() + self._task = asyncio.create_task(self._flush_loop()) + logger.debug("AsyncLogHiveLogger started for project '%s'", self.project) + + async def stop(self, flush: bool = True): + """Stop the background task.""" + self._stop_event.set() + if flush: + await self._flush_now() + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + if self._client: + await self._client.aclose() + logger.debug("AsyncLogHiveLogger stopped for project '%s'", self.project) + + async def __aenter__(self): + await self.start() + return self + + async def __aexit__(self, *args): + await self.stop() + + async def _flush_loop(self): + """Background loop that periodically flushes the queue.""" + while not self._stop_event.is_set(): + await self._flush_now() + await asyncio.sleep(self.flush_interval) + + async def _flush_now(self): + """Flush all currently queued entries.""" + entries = [] + while len(entries) < self.batch_size: + try: + entry = self._queue.get_nowait() + entries.append(entry) + except asyncio.QueueEmpty: + break + + if not entries: + return + + await self._send_batch(entries) + + async def _send_batch(self, entries: List[Dict[str, Any]]): + """Send a batch with retries.""" + url = urljoin(self.endpoint, "/api/logs/ingest") + payload = {"project": self.project, "entries": entries} + + for attempt in range(self.max_retries): + try: + resp = await self._client.post( + url, + json=payload, + headers={"Authorization": f"Bearer {self.api_key}"}, + ) + 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: + await asyncio.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.""" + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "level": level, + "message": message, + "logger": kwargs.pop("logger", None) or __name__, + "extra": kwargs, + } + + 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 ───────────────────────────────────────────── + + async def debug(self, message: str, **kwargs): + self._enqueue("debug", message, **kwargs) + + async def info(self, message: str, **kwargs): + self._enqueue("info", message, **kwargs) + + async def warning(self, message: str, **kwargs): + self._enqueue("warning", message, **kwargs) + + async def error(self, message: str, **kwargs): + self._enqueue("error", message, **kwargs) + + async def critical(self, message: str, **kwargs): + self._enqueue("critical", message, **kwargs) diff --git a/vendor/loghive_client/client.py b/vendor/loghive_client/client.py new file mode 100644 index 0000000..89937c8 --- /dev/null +++ b/vendor/loghive_client/client.py @@ -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) diff --git a/vendor/loghive_client/handler.py b/vendor/loghive_client/handler.py new file mode 100644 index 0000000..d45b225 --- /dev/null +++ b/vendor/loghive_client/handler.py @@ -0,0 +1,72 @@ +"""Python logging.Handler integration — use LogHive with the stdlib logging module. + +This allows you to replace or augment your existing logging setup with +zero code changes (just add a handler to your logger). +""" + +import logging +from typing import Optional + +from loghive_client.client import LogHiveLogger + + +class LogHiveHandler(logging.Handler): + """A logging.Handler that sends records to LogHive. + + Use it with Python's standard logging module: + import logging + from loghive_client import LogHiveHandler + + handler = LogHiveHandler("my-project", "api-key", "http://localhost:8000") + logging.getLogger().addHandler(handler) + + All existing logger calls (logger.info, logger.error, etc.) will + automatically forward to LogHive. + """ + + LEVEL_MAP = { + logging.DEBUG: "debug", + logging.INFO: "info", + logging.WARNING: "warning", + logging.ERROR: "error", + logging.CRITICAL: "critical", + } + + def __init__( + self, + project: str, + api_key: str, + endpoint: str = "http://localhost:8000", + level: int = logging.INFO, + ): + super().__init__(level=level) + self._client = LogHiveLogger( + project=project, + api_key=api_key, + endpoint=endpoint, + ) + + def emit(self, record: logging.LogRecord): + """Send a log record to LogHive.""" + try: + level = self.LEVEL_MAP.get(record.levelno, "info") + extra = { + "logger": record.name, + "module": record.module, + "function": record.funcName, + "line_no": record.lineno, + } + if record.exc_info and record.exc_info[0]: + import traceback + extra["exception"] = "".join( + traceback.format_exception(*record.exc_info) + ) + + self._client._enqueue(level, record.getMessage(), **extra) + except Exception: + self.handleError(record) + + def close(self): + """Flush and close.""" + self._client.stop(flush=True) + super().close() diff --git a/vendor/setup.py b/vendor/setup.py new file mode 100644 index 0000000..8fbf159 --- /dev/null +++ b/vendor/setup.py @@ -0,0 +1,23 @@ +"""Setup script for loghive-client.""" + +from setuptools import setup, find_packages + +setup( + name="loghive-client", + version="0.1.0", + description="LogHive client SDK — push logs from your Python projects to LogHive", + author="LogHive", + packages=find_packages(), + install_requires=[ + "httpx>=0.27.0", + ], + python_requires=">=3.10", + classifiers=[ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + ], +)