GX-gp-notify/vendor/loghive_client/client.py

"""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)