From abfd07331ee608ca2e165c30f3512003485b330d Mon Sep 17 00:00:00 2001 From: v6ole Date: Sat, 9 May 2026 14:55:14 +0800 Subject: [PATCH] Initial commit: LogHive centralized log management system Co-Authored-By: Claude Opus 4.7 --- .claude/settings.local.json | 22 ++ .env.example | 27 +++ .gitignore | 27 +++ README.md | 219 +++++++++++++++++ backend/Dockerfile | 21 ++ backend/app/__init__.py | 0 backend/app/api/__init__.py | 0 backend/app/api/alerts.py | 180 ++++++++++++++ backend/app/api/dashboard.py | 34 +++ backend/app/api/logs.py | 253 ++++++++++++++++++++ backend/app/api/projects.py | 123 ++++++++++ backend/app/config.py | 79 +++++++ backend/app/core/__init__.py | 1 + backend/app/core/auth.py | 50 ++++ backend/app/core/dependencies.py | 1 + backend/app/database.py | 37 +++ backend/app/main.py | 65 ++++++ backend/app/models/__init__.py | 5 + backend/app/models/alert.py | 71 ++++++ backend/app/models/log.py | 37 +++ backend/app/models/project.py | 34 +++ backend/app/schemas/__init__.py | 35 +++ backend/app/schemas/alert.py | 63 +++++ backend/app/schemas/log.py | 64 +++++ backend/app/schemas/project.py | 34 +++ backend/app/services/__init__.py | 0 backend/app/services/alert_service.py | 142 +++++++++++ backend/app/services/analytics_service.py | 125 ++++++++++ backend/celery_worker/__init__.py | 0 backend/celery_worker/celery_app.py | 33 +++ backend/celery_worker/tasks.py | 82 +++++++ backend/requirements.txt | 25 ++ client/README.md | 84 +++++++ client/loghive_client/__init__.py | 20 ++ client/loghive_client/async_client.py | 178 ++++++++++++++ client/loghive_client/client.py | 183 +++++++++++++++ client/loghive_client/handler.py | 72 ++++++ client/setup.py | 23 ++ docker-compose.yml | 42 ++++ frontend/Dockerfile | 20 ++ frontend/index.html | 13 ++ frontend/nginx.conf | 21 ++ frontend/package.json | 25 ++ frontend/public/favicon.svg | 4 + frontend/src/App.vue | 69 ++++++ frontend/src/api/index.js | 92 ++++++++ frontend/src/main.js | 20 ++ frontend/src/router/index.js | 19 ++ frontend/src/views/AlertManageView.vue | 272 ++++++++++++++++++++++ frontend/src/views/DashboardView.vue | 204 ++++++++++++++++ frontend/src/views/LogSearchView.vue | 222 ++++++++++++++++++ frontend/src/views/ProjectManageView.vue | 179 ++++++++++++++ frontend/vite.config.js | 15 ++ 接入LogHive日志系统.md | 150 ++++++++++++ 54 files changed, 3816 insertions(+) create mode 100644 .claude/settings.local.json create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 backend/Dockerfile create mode 100644 backend/app/__init__.py create mode 100644 backend/app/api/__init__.py create mode 100644 backend/app/api/alerts.py create mode 100644 backend/app/api/dashboard.py create mode 100644 backend/app/api/logs.py create mode 100644 backend/app/api/projects.py create mode 100644 backend/app/config.py create mode 100644 backend/app/core/__init__.py create mode 100644 backend/app/core/auth.py create mode 100644 backend/app/core/dependencies.py create mode 100644 backend/app/database.py create mode 100644 backend/app/main.py create mode 100644 backend/app/models/__init__.py create mode 100644 backend/app/models/alert.py create mode 100644 backend/app/models/log.py create mode 100644 backend/app/models/project.py create mode 100644 backend/app/schemas/__init__.py create mode 100644 backend/app/schemas/alert.py create mode 100644 backend/app/schemas/log.py create mode 100644 backend/app/schemas/project.py create mode 100644 backend/app/services/__init__.py create mode 100644 backend/app/services/alert_service.py create mode 100644 backend/app/services/analytics_service.py create mode 100644 backend/celery_worker/__init__.py create mode 100644 backend/celery_worker/celery_app.py create mode 100644 backend/celery_worker/tasks.py create mode 100644 backend/requirements.txt create mode 100644 client/README.md create mode 100644 client/loghive_client/__init__.py create mode 100644 client/loghive_client/async_client.py create mode 100644 client/loghive_client/client.py create mode 100644 client/loghive_client/handler.py create mode 100644 client/setup.py create mode 100644 docker-compose.yml create mode 100644 frontend/Dockerfile create mode 100644 frontend/index.html create mode 100644 frontend/nginx.conf create mode 100644 frontend/package.json create mode 100644 frontend/public/favicon.svg create mode 100644 frontend/src/App.vue create mode 100644 frontend/src/api/index.js create mode 100644 frontend/src/main.js create mode 100644 frontend/src/router/index.js create mode 100644 frontend/src/views/AlertManageView.vue create mode 100644 frontend/src/views/DashboardView.vue create mode 100644 frontend/src/views/LogSearchView.vue create mode 100644 frontend/src/views/ProjectManageView.vue create mode 100644 frontend/vite.config.js create mode 100644 接入LogHive日志系统.md diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 0000000..c70bb83 --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,22 @@ +{ + "permissions": { + "allow": [ + "Bash(docker --version)", + "Bash(docker compose *)", + "Bash(curl *)", + "Bash(docker cp *)", + "Bash(docker stop *)", + "Bash(docker rm *)", + "Bash(docker network *)", + "Bash(docker volume *)", + "Bash(python -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\('timestamp:', d['hits'][0]['timestamp']\\)\")", + "Bash(python -c ' *)", + "Bash(python -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\('bucket:', d['time_buckets'][0]['key'] if d['time_buckets'] else 'none'\\)\")", + "Bash(python -c \"import sys,json; d=json.load\\(sys.stdin\\); print\\('found in range:', d['total'], 'logs'\\)\")", + "Bash(git init *)", + "Bash(git remote *)", + "Bash(git add *)", + "Bash(git config *)" + ] + } +} diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..2457caf --- /dev/null +++ b/.env.example @@ -0,0 +1,27 @@ +# LogHive Environment Configuration + +# ── API ───────────────────────────────────────────────────── +API_HOST=0.0.0.0 +API_PORT=8000 +DEBUG=false +CORS_ORIGINS=* + +# ── Security ──────────────────────────────────────────────── +SECRET_KEY=change-me-to-a-random-string +TOKEN_EXPIRE_HOURS=720 + +# ── PostgreSQL ────────────────────────────────────────────── +DATABASE_URL=postgresql+asyncpg://LogHive:LogHive@localhost:5432/LogHive + +# ── Redis ─────────────────────────────────────────────────── +REDIS_URL=redis://localhost:6379/0 +CELERY_BROKER_URL=redis://localhost:6379/1 +CELERY_RESULT_BACKEND=redis://localhost:6379/2 + +# ── Log Retention ─────────────────────────────────────────── +LOG_RETENTION_DAYS=30 +LOG_COLD_STORAGE_DAYS=7 + +# ── Notifications (optional) ──────────────────────────────── +FEISHU_WEBHOOK_URL= +DINGTALK_WEBHOOK_URL= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..187cb6d --- /dev/null +++ b/.gitignore @@ -0,0 +1,27 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.venv/ +venv/ + +# Environment +.env + +# Node / Frontend +node_modules/ +frontend/dist/ + +# IDE +.idea/ +.vscode/ +*.swp + +# Docker +docker-compose.override.yml + +# OS +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..eb508d0 --- /dev/null +++ b/README.md @@ -0,0 +1,219 @@ +# 🐝 LogHive — 集中式日志管理系统 + +LogHive 是一个集中式的日志管理系统，可以从你的多个 Python 项目中收集日志，并提供搜索、分析和告警能力。 + +## 架构概览 + +``` +┌──────────────┐ ┌──────────────┐ ┌──────────────┐ +│ Python 项目 │ │ Python 项目 │ │ Python 项目 │ +│ (LogHive SDK)│ │ (LogHive SDK)│ │ (LogHive SDK)│ +└──────┬───────┘ └──────┬───────┘ └──────┬───────┘ + │ │ │ + └───────────────────┼───────────────────┘ + │ HTTPS / REST API + ▼ + ┌─────────────────┐ + │ LogHive Backend │ + │ (FastAPI) │ + └────────┬─────────┘ + │ + ┌────────────┼────────────┐ + ▼ ▼ ▼ + ┌──────────┐ ┌──────────┐ ┌──────────┐ + │ ES │ │PostgreSQL│ │ Redis │ + │ 日志存储 │ │ 元数据 │ │ 队列/缓存 │ + └──────────┘ └──────────┘ └──────────┘ + │ + ┌──────┴──────┐ + ▼ ▼ + ┌─────────┐ ┌──────────┐ + │ Celery │ │ Vue 3 │ + │ 异步任务 │ │ 前端 │ + └─────────┘ └──────────┘ +``` + +## 功能特性 + +- **📥 日志收集** — 通过 REST API 接收 JSON 结构化日志，支持同步/异步客户端 +- **🔍 全文检索** — 基于 Elasticsearch，支持按项目、级别、关键词、Trace ID、时间范围过滤 +- **📊 统计分析** — 错误率趋势、级别分布、TOP N 错误、时序图表 +- **🔔 告警规则** — 可配置的阈值告警，支持飞书/钉钉通知 +- **🔗 链路追踪** — 通过 Trace ID 关联同一请求的跨模块日志 +- **🐍 Python SDK** — 一行代码集成到现有项目，支持标准 logging handler +- **🔄 前后端分离** — FastAPI 后端 + Vue 3 前端 + +## 快速启动 + +### 前置条件 + +- Docker & Docker Compose +- Python 3.10+（本地开发） + +### 使用 Docker Compose（推荐） + +```bash +# 克隆项目 +cd LogHive + +# 启动所有服务 +docker compose up -d + +# 检查状态 +docker compose ps +``` + +启动后访问： +- **前端**：http://localhost:3000 +- **API**：http://localhost:8000 +- **API 健康检查**：http://localhost:8000/api/health +- **Kibana**：http://localhost:5601 + +### 本地开发 + +```bash +# 1. 启动依赖服务 +docker compose up -d postgres elasticsearch redis + +# 2. 启动后端 +cd backend +pip install -r requirements.txt +uvicorn app.main:app --reload --port 8000 + +# 3. 启动前端 +cd frontend +npm install +npm run dev +``` + +## 项目结构 + +``` +LogHive/ +├── backend/ # FastAPI 后端 +│ ├── app/ +│ │ ├── api/ # API 路由 +│ │ │ ├── logs.py # 日志收录 & 查询 +│ │ │ ├── projects.py # 项目管理 +│ │ │ ├── alerts.py # 告警管理 +│ │ │ └── dashboard.py # 仪表盘 +│ │ ├── models/ # SQLAlchemy 模型 +│ │ ├── schemas/ # Pydantic 数据模型 +│ │ ├── services/ # 业务逻辑层 +│ │ ├── core/ # 认证 & 工具 +│ │ ├── config.py # 配置 +│ │ ├── database.py # 数据库连接 +│ │ └── main.py # 入口 +│ ├── celery_worker/ # Celery 异步任务 +│ ├── requirements.txt +│ └── Dockerfile +├── client/ # Python SDK +│ ├── loghive_client/ +│ │ ├── client.py # 同步客户端 +│ │ ├── async_client.py # 异步客户端 +│ │ └── handler.py # logging 集成 +│ ├── setup.py +│ └── README.md +├── frontend/ # Vue 3 前端 +│ ├── src/ +│ │ ├── api/ # API 调用层 +│ │ ├── router/ # 路由 +│ │ └── views/ # 页面 +│ ├── package.json +│ ├── vite.config.js +│ └── Dockerfile +├── docker-compose.yml +├── .env.example +└── README.md +``` + +## 使用指南 + +### 1. 创建项目 & 获取 API Key + +```bash +# 注册项目 +curl -X POST http://localhost:8000/api/projects \ + -H "Content-Type: application/json" \ + -d '{"name": "my-app", "description": "我的 Python 应用"}' + +# 获取 API Key（在响应的 api_key 字段） +``` + +### 2. 在你的项目中使用 SDK + +```python +# 安装 SDK +# pip install loghive-client + +from loghive_client import LogHiveLogger + +logger = LogHiveLogger( + project="my-app", + api_key="your-api-key", + endpoint="http://localhost:8000", +) + +logger.info("服务启动成功", extra={"port": 8080}) +logger.error("数据库连接超时", exc_info=True) +``` + +### 3. 标准 logging 集成（零代码改动） + +```python +import logging +from loghive_client import LogHiveHandler + +handler = LogHiveHandler("my-app", "api-key", "http://localhost:8000") +logging.getLogger().addHandler(handler) + +# 所有现有的日志调用自动转发到 LogHive +logging.info("这条日志也会发送到 LogHive") +``` + +### 4. 直接通过 API 推送日志 + +```bash +curl -X POST http://localhost:8000/api/logs/ingest \ + -H "Authorization: Bearer your-api-key" \ + -H "Content-Type: application/json" \ + -d '{ + "project": "my-app", + "entries": [ + { + "level": "info", + "message": "Hello from my app!", + "logger": "myapp.server", + "extra": {"user_id": 42} + } + ] + }' +``` + +## 技术栈 + +| 层级 | 技术 | +|------|------| +| 后端框架 | FastAPI (Python 3.12) | +| 日志存储 | Elasticsearch 8.x | +| 元数据存储 | PostgreSQL 16 / TimescaleDB | +| 消息队列 | Redis 7 | +| 异步任务 | Celery | +| 前端 | Vue 3 + Element Plus + ECharts | +| 部署 | Docker Compose | + +## 配置 + +通过环境变量配置，详见 `.env.example`。主要配置项： + +| 变量 | 默认值 | 说明 | +|------|--------|------| +| `SECRET_KEY` | — | JWT 密钥（生产必改） | +| `ES_HOST` | localhost | Elasticsearch 地址 | +| `DATABASE_URL` | — | PostgreSQL 连接串 | +| `LOG_RETENTION_DAYS` | 30 | 日志保留天数 | +| `FEISHU_WEBHOOK_URL` | — | 飞书机器人 Webhook | + +## License + +MIT diff --git a/backend/Dockerfile b/backend/Dockerfile new file mode 100644 index 0000000..abad861 --- /dev/null +++ b/backend/Dockerfile @@ -0,0 +1,21 @@ +FROM python:3.12-slim + +WORKDIR /app + +# Install system dependencies +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# Install Python dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application +COPY . . + +# Expose port +EXPOSE 8000 + +# Run with uvicorn +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/backend/app/__init__.py b/backend/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/api/alerts.py b/backend/app/api/alerts.py new file mode 100644 index 0000000..c706e0c --- /dev/null +++ b/backend/app/api/alerts.py @@ -0,0 +1,180 @@ +"""Alert management API routes.""" + +from datetime import datetime, timezone +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models.alert import AlertRule, AlertHistory, AlertLevel, AlertOperator +from app.schemas.alert import ( + AlertRuleCreate, + AlertRuleUpdate, + AlertRuleResponse, + AlertHistoryResponse, +) + +router = APIRouter(prefix="/api/alerts", tags=["alerts"]) + + +# ── Rule CRUD ────────────────────────────────────────────────── + + +@router.post("/rules", response_model=AlertRuleResponse, status_code=201) +async def create_alert_rule( + data: AlertRuleCreate, + db: AsyncSession = Depends(get_db), +): + """Create a new alert rule.""" + rule = AlertRule( + project_id=data.project_id, + name=data.name, + level=data.level, + field=data.field, + operator=data.operator, + threshold=data.threshold, + window_minutes=data.window_minutes, + notify_channels=",".join(data.notify_channels) if data.notify_channels else "", + ) + db.add(rule) + await db.flush() + await db.refresh(rule) + return _rule_to_response(rule) + + +@router.get("/rules", response_model=List[AlertRuleResponse]) +async def list_alert_rules( + project_id: Optional[str] = Query(None), + db: AsyncSession = Depends(get_db), +): + """List alert rules, optionally filtered by project.""" + stmt = select(AlertRule) + if project_id: + stmt = stmt.where(AlertRule.project_id == project_id) + stmt = stmt.order_by(AlertRule.created_at.desc()) + result = await db.execute(stmt) + return [_rule_to_response(r) for r in result.scalars().all()] + + +@router.get("/rules/{rule_id}", response_model=AlertRuleResponse) +async def get_alert_rule(rule_id: str, db: AsyncSession = Depends(get_db)): + """Get a single alert rule.""" + result = await db.execute(select(AlertRule).where(AlertRule.id == rule_id)) + rule = result.scalar_one_or_none() + if not rule: + raise HTTPException(status_code=404, detail="Alert rule not found") + return _rule_to_response(rule) + + +@router.patch("/rules/{rule_id}", response_model=AlertRuleResponse) +async def update_alert_rule( + rule_id: str, + data: AlertRuleUpdate, + db: AsyncSession = Depends(get_db), +): + """Update an alert rule.""" + result = await db.execute(select(AlertRule).where(AlertRule.id == rule_id)) + rule = result.scalar_one_or_none() + if not rule: + raise HTTPException(status_code=404, detail="Alert rule not found") + + if data.name is not None: + rule.name = data.name + if data.level is not None: + rule.level = data.level + if data.field is not None: + rule.field = data.field + if data.operator is not None: + rule.operator = data.operator + if data.threshold is not None: + rule.threshold = data.threshold + if data.window_minutes is not None: + rule.window_minutes = data.window_minutes + if data.is_enabled is not None: + rule.is_enabled = data.is_enabled + if data.notify_channels is not None: + rule.notify_channels = ",".join(data.notify_channels) + + await db.flush() + await db.refresh(rule) + return _rule_to_response(rule) + + +@router.delete("/rules/{rule_id}", status_code=204) +async def delete_alert_rule(rule_id: str, db: AsyncSession = Depends(get_db)): + """Delete an alert rule.""" + result = await db.execute(select(AlertRule).where(AlertRule.id == rule_id)) + rule = result.scalar_one_or_none() + if not rule: + raise HTTPException(status_code=404, detail="Alert rule not found") + await db.delete(rule) + + +# ── Alert History ────────────────────────────────────────────── + + +@router.get("/history", response_model=List[AlertHistoryResponse]) +async def list_alert_history( + project_id: Optional[str] = Query(None), + rule_id: Optional[str] = Query(None), + limit: int = Query(100, ge=1, le=1000), + db: AsyncSession = Depends(get_db), +): + """List alert trigger history.""" + stmt = select(AlertHistory).order_by(AlertHistory.triggered_at.desc()).limit(limit) + if project_id: + stmt = stmt.where(AlertHistory.project_id == project_id) + if rule_id: + stmt = stmt.where(AlertHistory.rule_id == rule_id) + result = await db.execute(stmt) + return [_history_to_response(h) for h in result.scalars().all()] + + +@router.post("/history/{history_id}/acknowledge", response_model=AlertHistoryResponse) +async def acknowledge_alert( + history_id: str, + db: AsyncSession = Depends(get_db), +): + """Mark an alert as acknowledged.""" + result = await db.execute(select(AlertHistory).where(AlertHistory.id == history_id)) + history = result.scalar_one_or_none() + if not history: + raise HTTPException(status_code=404, detail="Alert history not found") + history.is_acknowledged = True + await db.flush() + await db.refresh(history) + return _history_to_response(history) + + +# ── Helpers ──────────────────────────────────────────────────── + + +def _rule_to_response(rule: AlertRule) -> AlertRuleResponse: + return AlertRuleResponse( + id=rule.id, + project_id=rule.project_id, + name=rule.name, + level=rule.level.value if hasattr(rule.level, "value") else rule.level, + field=rule.field, + operator=rule.operator.value if hasattr(rule.operator, "value") else rule.operator, + threshold=rule.threshold, + window_minutes=rule.window_minutes, + is_enabled=rule.is_enabled, + notify_channels=rule.notify_channels.split(",") if rule.notify_channels else [], + created_at=rule.created_at, + ) + + +def _history_to_response(history: AlertHistory) -> AlertHistoryResponse: + return AlertHistoryResponse( + id=history.id, + rule_id=history.rule_id, + project_id=history.project_id, + level=history.level.value if hasattr(history.level, "value") else history.level, + message=history.message, + triggered_value=history.triggered_value, + is_acknowledged=history.is_acknowledged, + triggered_at=history.triggered_at, + ) diff --git a/backend/app/api/dashboard.py b/backend/app/api/dashboard.py new file mode 100644 index 0000000..3d6564e --- /dev/null +++ b/backend/app/api/dashboard.py @@ -0,0 +1,34 @@ +"""Dashboard API routes.""" + +from typing import Optional + +from fastapi import APIRouter, Depends, Query +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.services.analytics_service import AnalyticsService + +router = APIRouter(prefix="/api/dashboard", tags=["dashboard"]) + + +@router.get("/stats") +async def dashboard_stats( + project: Optional[str] = Query(None), + hours: int = Query(24, ge=1, le=720), + db: AsyncSession = Depends(get_db), +): + """Get overview statistics for the dashboard.""" + service = AnalyticsService(db) + return await service.get_dashboard_stats(project=project, hours=hours) + + +@router.get("/timeseries") +async def dashboard_timeseries( + project: Optional[str] = Query(None), + hours: int = Query(24, ge=1, le=720), + interval: str = Query("5m", max_length=16), + db: AsyncSession = Depends(get_db), +): + """Get time-series data for chart rendering.""" + service = AnalyticsService(db) + return await service.get_time_series(project=project, hours=hours, interval=interval) diff --git a/backend/app/api/logs.py b/backend/app/api/logs.py new file mode 100644 index 0000000..b731236 --- /dev/null +++ b/backend/app/api/logs.py @@ -0,0 +1,253 @@ +"""Log ingestion and query API routes.""" + +import logging +from datetime import datetime, timezone, timedelta +from typing import Optional + +from fastapi import APIRouter, Depends, Header, HTTPException, Query, status +from sqlalchemy import func, select, delete, and_ +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings, to_display_time, from_query_time +from app.database import get_db +from app.schemas.log import ( + LogBatchRequest, + LogEntry as LogEntrySchema, + LogQueryParams, + LogSearchResult, + LogStatsResult, +) +from app.models.project import Project +from app.models.log import LogEntry +from app.core.auth import verify_api_key, resolve_project_by_api_key + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/logs", tags=["logs"]) + + +# ── Ingestion ───────────────────────────────────────────────── + + +@router.post("/ingest", status_code=201) +async def ingest_logs( + batch: LogBatchRequest, + authorization: Optional[str] = Header(None), + db: AsyncSession = Depends(get_db), +): + """Receive a batch of log entries from a project.""" + api_key = await verify_api_key(authorization) + project = await resolve_project_by_api_key(db, api_key) + + now = datetime.now(timezone.utc) + rows = [] + for entry in batch.entries: + row = LogEntry( + project_id=project.id, + project_name=project.name, + timestamp=entry.timestamp or now, + level=entry.level, + logger=entry.logger, + message=entry.message, + module=entry.module, + function=entry.function, + line_no=entry.line_no, + trace_id=entry.trace_id, + exception=entry.exception, + extra=entry.extra, + ) + rows.append(row) + + db.add_all(rows) + await db.flush() + return {"accepted": len(rows), "project": project.name} + + +@router.post("/ingest/single", status_code=201) +async def ingest_single_log( + entry: LogEntrySchema, + project_name: str = Query(..., description="Project name"), + authorization: Optional[str] = Header(None), + db: AsyncSession = Depends(get_db), +): + """Receive a single log entry (useful for testing / simple clients).""" + api_key = await verify_api_key(authorization) + project = await resolve_project_by_api_key(db, api_key) + + now = datetime.now(timezone.utc) + row = LogEntry( + project_id=project.id, + project_name=project.name, + timestamp=entry.timestamp or now, + level=entry.level, + logger=entry.logger, + message=entry.message, + module=entry.module, + function=entry.function, + line_no=entry.line_no, + trace_id=entry.trace_id, + exception=entry.exception, + extra=entry.extra, + ) + db.add(row) + await db.flush() + return {"accepted": 1, "project": project.name} + + +# ── Query ────────────────────────────────────────────────────── + + +@router.get("/search", response_model=LogSearchResult) +async def search_logs( + project: Optional[str] = Query(None, description="Filter by project name"), + level: Optional[str] = Query(None, pattern="^(debug|info|warning|error|critical)$"), + query: Optional[str] = Query(None, max_length=1024), + trace_id: Optional[str] = Query(None, max_length=64), + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + page: int = Query(1, ge=1), + page_size: int = Query(50, ge=1, le=1000), + sort_by: str = Query("timestamp", max_length=32), + sort_order: str = Query("desc", pattern="^(asc|desc)$"), + db: AsyncSession = Depends(get_db), +): + """Search logs with full-text query and filters.""" + conditions = [] + + if project: + conditions.append(LogEntry.project_name == project) + if level: + conditions.append(LogEntry.level == level) + if trace_id: + conditions.append(LogEntry.trace_id == trace_id) + if start_time: + conditions.append(LogEntry.timestamp >= from_query_time(start_time)) + if end_time: + conditions.append(LogEntry.timestamp <= from_query_time(end_time)) + if query: + conditions.append( + func.to_tsvector( + "english", + func.concat( + LogEntry.message, " ", + func.coalesce(LogEntry.logger, ""), " ", + func.coalesce(LogEntry.module, ""), + ), + ).bool_op("@@")( + func.plainto_tsquery("english", query) + ) + ) + + # Sort + sort_col = getattr(LogEntry, sort_by, LogEntry.timestamp) + order = sort_col.desc() if sort_order == "desc" else sort_col.asc() + + # Count total + count_stmt = select(func.count(LogEntry.id)) + if conditions: + count_stmt = count_stmt.where(and_(*conditions)) + total = (await db.execute(count_stmt)).scalar() or 0 + + # Fetch page + stmt = select(LogEntry).order_by(order).offset((page - 1) * page_size).limit(page_size) + if conditions: + stmt = stmt.where(and_(*conditions)) + rows = (await db.execute(stmt)).scalars().all() + + hits = [] + for r in rows: + hits.append({ + "project_id": r.project_id, + "project_name": r.project_name, + "timestamp": to_display_time(r.timestamp), + "level": r.level, + "logger": r.logger, + "message": r.message, + "module": r.module, + "function": r.function, + "line_no": r.line_no, + "trace_id": r.trace_id, + "exception": r.exception, + "extra": r.extra, + }) + + return LogSearchResult(total=total, page=page, page_size=page_size, hits=hits) + + +@router.get("/stats", response_model=LogStatsResult) +async def log_stats( + project: Optional[str] = Query(None), + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + interval: str = Query("5m", max_length=16), + db: AsyncSession = Depends(get_db), +): + """Get aggregated log statistics over a time range.""" + conditions = [] + if project: + conditions.append(LogEntry.project_name == project) + if start_time: + conditions.append(LogEntry.timestamp >= from_query_time(start_time)) + if end_time: + conditions.append(LogEntry.timestamp <= from_query_time(end_time)) + + def apply(stmt): + if conditions: + stmt = stmt.where(and_(*conditions)) + return stmt + + # Time buckets (hourly) + stmt = apply( + select( + func.date_trunc("hour", LogEntry.timestamp).label("bucket"), + func.count(LogEntry.id).label("cnt"), + ).group_by("bucket").order_by("bucket") + ) + time_rows = (await db.execute(stmt)).all() + time_buckets = [ + {"key": to_display_time(r.bucket), "doc_count": r.cnt} + for r in time_rows + ] + + # Level counts + stmt = apply( + select(LogEntry.level, func.count(LogEntry.id).label("cnt")) + .group_by(LogEntry.level) + ) + level_rows = (await db.execute(stmt)).all() + level_counts = {r.level: r.cnt for r in level_rows} + + # Top loggers + stmt = apply( + select(LogEntry.logger, func.count(LogEntry.id).label("cnt")) + .group_by(LogEntry.logger) + .order_by(func.count(LogEntry.id).desc()) + .limit(20) + ) + logger_rows = (await db.execute(stmt)).all() + top_loggers = [{"logger": r.logger, "count": r.cnt} for r in logger_rows] + + # Total count + total = (await db.execute(apply(select(func.count(LogEntry.id))))).scalar() or 0 + + return LogStatsResult( + project=project, + time_buckets=time_buckets, + level_counts=level_counts, + top_loggers=top_loggers, + total=total, + ) + + +@router.delete("/cleanup") +async def cleanup_old_logs( + db: AsyncSession = Depends(get_db), +): + """Delete log entries older than the retention period.""" + cutoff = datetime.now(timezone.utc) - timedelta(days=settings.LOG_RETENTION_DAYS) + + result = await db.execute( + delete(LogEntry).where(LogEntry.timestamp < cutoff) + ) + await db.flush() + + return {"deleted_count": result.rowcount} diff --git a/backend/app/api/projects.py b/backend/app/api/projects.py new file mode 100644 index 0000000..4655970 --- /dev/null +++ b/backend/app/api/projects.py @@ -0,0 +1,123 @@ +"""Project management API routes.""" + +import uuid +from datetime import datetime, timezone +from typing import List + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models.project import Project +from app.schemas.project import ProjectCreate, ProjectUpdate, ProjectResponse, ProjectWithToken + +router = APIRouter(prefix="/api/projects", tags=["projects"]) + + +@router.post("", response_model=ProjectWithToken, status_code=201) +async def create_project( + data: ProjectCreate, + db: AsyncSession = Depends(get_db), +): + """Register a new project for log collection.""" + existing = await db.execute(select(Project).where(Project.name == data.name)) + if existing.scalar_one_or_none(): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Project '{data.name}' already exists", + ) + + project = Project( + name=data.name, + description=data.description, + api_key=uuid.uuid4().hex, + ) + db.add(project) + await db.flush() + await db.refresh(project) + return project + + +@router.get("", response_model=List[ProjectResponse]) +async def list_projects( + db: AsyncSession = Depends(get_db), +): + """List all registered projects.""" + result = await db.execute(select(Project).order_by(Project.created_at.desc())) + return result.scalars().all() + + +@router.get("/{project_id}", response_model=ProjectResponse) +async def get_project( + project_id: str, + db: AsyncSession = Depends(get_db), +): + """Get a single project by ID.""" + result = await db.execute(select(Project).where(Project.id == project_id)) + project = result.scalar_one_or_none() + if not project: + raise HTTPException(status_code=404, detail="Project not found") + return project + + +@router.patch("/{project_id}", response_model=ProjectResponse) +async def update_project( + project_id: str, + data: ProjectUpdate, + db: AsyncSession = Depends(get_db), +): + """Update a project.""" + result = await db.execute(select(Project).where(Project.id == project_id)) + project = result.scalar_one_or_none() + if not project: + raise HTTPException(status_code=404, detail="Project not found") + + if data.name is not None: + # Check uniqueness + existing = await db.execute( + select(Project).where(Project.name == data.name, Project.id != project_id) + ) + if existing.scalar_one_or_none(): + raise HTTPException(status_code=409, detail="Name already taken") + project.name = data.name + if data.description is not None: + project.description = data.description + if data.is_active is not None: + project.is_active = data.is_active + + project.updated_at = datetime.now(timezone.utc) + await db.flush() + await db.refresh(project) + return project + + +@router.delete("/{project_id}", status_code=204) +async def delete_project( + project_id: str, + db: AsyncSession = Depends(get_db), +): + """Delete a project.""" + result = await db.execute(select(Project).where(Project.id == project_id)) + project = result.scalar_one_or_none() + if not project: + raise HTTPException(status_code=404, detail="Project not found") + await db.delete(project) + + +@router.post("/{project_id}/rotate-key", response_model=ProjectWithToken) +async def rotate_api_key( + project_id: str, + db: AsyncSession = Depends(get_db), +): + """Rotate a project's API key.""" + result = await db.execute(select(Project).where(Project.id == project_id)) + project = result.scalar_one_or_none() + if not project: + raise HTTPException(status_code=404, detail="Project not found") + + project.api_key = uuid.uuid4().hex + project.updated_at = datetime.now(timezone.utc) + await db.flush() + await db.refresh(project) + return project diff --git a/backend/app/config.py b/backend/app/config.py new file mode 100644 index 0000000..b43ecb4 --- /dev/null +++ b/backend/app/config.py @@ -0,0 +1,79 @@ +"""LogHive configuration.""" + +import os +from datetime import datetime, timezone, timedelta +from typing import Optional + +# Beijing timezone offset +_BEIJING_TZ = timezone(timedelta(hours=8)) + + +def to_display_time(dt: datetime) -> str: + """Convert a UTC datetime to Beijing time for display. + Returns ISO format without timezone suffix (e.g. '2026-05-08T14:30:00').""" + if dt is None: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + local = dt.astimezone(_BEIJING_TZ) + return local.strftime("%Y-%m-%dT%H:%M:%S") + + +def from_query_time(dt: Optional[datetime]) -> Optional[datetime]: + """Interpret a naive datetime query param as Beijing time, return UTC datetime.""" + if dt is None: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=_BEIJING_TZ) + return dt.astimezone(timezone.utc) + + +class Settings: + """Application settings loaded from environment variables.""" + + # Service name + SERVICE_NAME: str = "LogHive" + + # API + API_HOST: str = os.getenv("API_HOST", "0.0.0.0") + API_PORT: int = int(os.getenv("API_PORT", "8000")) + DEBUG: bool = os.getenv("DEBUG", "false").lower() == "true" + CORS_ORIGINS: list[str] = os.getenv("CORS_ORIGINS", "*").split(",") + + # Security + SECRET_KEY: str = os.getenv("SECRET_KEY", "change-me-in-production") + TOKEN_EXPIRE_HOURS: int = int(os.getenv("TOKEN_EXPIRE_HOURS", "720")) + + # PostgreSQL + DATABASE_URL: str = os.getenv( + "DATABASE_URL", + "postgresql+asyncpg://LogHive:LogHive@localhost:5432/LogHive", + ) + + # Redis + REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0") + + # Celery + CELERY_BROKER_URL: str = os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/1") + CELERY_RESULT_BACKEND: str = os.getenv( + "CELERY_RESULT_BACKEND", "redis://localhost:6379/2" + ) + + # Log retention (days) + LOG_RETENTION_DAYS: int = int(os.getenv("LOG_RETENTION_DAYS", "30")) + LOG_COLD_STORAGE_DAYS: int = int(os.getenv("LOG_COLD_STORAGE_DAYS", "7")) + + # Rate limit + RATE_LIMIT_PER_SECOND: int = int(os.getenv("RATE_LIMIT_PER_SECOND", "1000")) + + # Alert + ALERT_CHECK_INTERVAL_SECONDS: int = int( + os.getenv("ALERT_CHECK_INTERVAL_SECONDS", "60") + ) + + # Notification + FEISHU_WEBHOOK_URL: Optional[str] = os.getenv("FEISHU_WEBHOOK_URL", None) + DINGTALK_WEBHOOK_URL: Optional[str] = os.getenv("DINGTALK_WEBHOOK_URL", None) + + +settings = Settings() diff --git a/backend/app/core/__init__.py b/backend/app/core/__init__.py new file mode 100644 index 0000000..969df0c --- /dev/null +++ b/backend/app/core/__init__.py @@ -0,0 +1 @@ +from .auth import verify_api_key, resolve_project_by_api_key diff --git a/backend/app/core/auth.py b/backend/app/core/auth.py new file mode 100644 index 0000000..4b6efb3 --- /dev/null +++ b/backend/app/core/auth.py @@ -0,0 +1,50 @@ +"""Auth utilities — API key verification.""" + +import hmac +from typing import Optional + +from fastapi import Header, HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models.project import Project + + +async def verify_api_key( + authorization: Optional[str] = Header(None), +) -> str: + """Extract and verify the API key from the Authorization header. + + Returns the project_id on success. + """ + if not authorization: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing Authorization header", + ) + + scheme, _, key = authorization.partition(" ") + if scheme.lower() != "bearer" or not key.strip(): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid Authorization format. Use: Bearer ", + ) + + return key.strip() + + +async def resolve_project_by_api_key( + db: AsyncSession, api_key: str +) -> Project: + """Look up a project by its API key.""" + result = await db.execute( + select(Project).where(Project.api_key == api_key, Project.is_active == True) + ) + project = result.scalar_one_or_none() + if not project: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid or inactive API key", + ) + return project diff --git a/backend/app/core/dependencies.py b/backend/app/core/dependencies.py new file mode 100644 index 0000000..f8fc1af --- /dev/null +++ b/backend/app/core/dependencies.py @@ -0,0 +1 @@ +"""Core auth logic.""" diff --git a/backend/app/database.py b/backend/app/database.py new file mode 100644 index 0000000..16922a6 --- /dev/null +++ b/backend/app/database.py @@ -0,0 +1,37 @@ +"""Database connections and session management.""" + +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker +from sqlalchemy.orm import DeclarativeBase + +from app.config import settings + +# ── PostgreSQL ──────────────────────────────────────────────────── +engine = create_async_engine(settings.DATABASE_URL, echo=settings.DEBUG) +async_session_factory = async_sessionmaker(engine, expire_on_commit=False) + + +class Base(DeclarativeBase): + pass + + +async def get_db() -> AsyncSession: + """Yield a DB session per request.""" + async with async_session_factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + finally: + await session.close() + + +async def init_db() -> None: + """Create all tables (use Alembic in production).""" + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + +async def close_db() -> None: + await engine.dispose() diff --git a/backend/app/main.py b/backend/app/main.py new file mode 100644 index 0000000..b84a658 --- /dev/null +++ b/backend/app/main.py @@ -0,0 +1,65 @@ +"""LogHive — Centralized Log Management System. + +FastAPI application that receives, stores, searches and analyzes logs +from multiple Python projects. +""" + +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware + +from app.config import settings +from app.database import init_db, close_db +from app.api import logs, projects, alerts, dashboard + +# ── Logging ──────────────────────────────────────────────────── +logging.basicConfig( + level=logging.INFO if not settings.DEBUG else logging.DEBUG, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", +) +logger = logging.getLogger(__name__) + + +# ── Lifecycle ────────────────────────────────────────────────── + + +@asynccontextmanager +async def lifespan(app: FastAPI): + """Startup and shutdown lifecycle.""" + logger.info("🚀 LogHive starting up...") + await init_db() + yield + logger.info("🛑 LogHive shutting down...") + await close_db() + + +# ── App ──────────────────────────────────────────────────────── + +app = FastAPI( + title="LogHive API", + description="Centralized log management system — ingest, search, monitor, and alert on logs from your Python services.", + version="0.1.0", + lifespan=lifespan, +) + +app.add_middleware( + CORSMiddleware, + allow_origins=settings.CORS_ORIGINS, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# ── Routes ───────────────────────────────────────────────────── + +app.include_router(logs.router) +app.include_router(projects.router) +app.include_router(alerts.router) +app.include_router(dashboard.router) + + +@app.get("/api/health") +async def health(): + return {"status": "ok", "service": settings.SERVICE_NAME, "version": "0.1.0"} diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py new file mode 100644 index 0000000..bfe8023 --- /dev/null +++ b/backend/app/models/__init__.py @@ -0,0 +1,5 @@ +from app.models.project import Project +from app.models.alert import AlertRule, AlertHistory +from app.models.log import LogEntry + +__all__ = ["Project", "AlertRule", "AlertHistory", "LogEntry"] diff --git a/backend/app/models/alert.py b/backend/app/models/alert.py new file mode 100644 index 0000000..27f74c5 --- /dev/null +++ b/backend/app/models/alert.py @@ -0,0 +1,71 @@ +"""Alert models — rules and history.""" + +import uuid +from datetime import datetime, timezone + +from sqlalchemy import String, Integer, Float, Boolean, DateTime, Text, Enum as SAEnum +from sqlalchemy.orm import Mapped, mapped_column +import enum + +from app.database import Base + + +class AlertLevel(str, enum.Enum): + INFO = "info" + WARNING = "warning" + CRITICAL = "critical" + + +class AlertOperator(str, enum.Enum): + GT = "gt" + GTE = "gte" + LT = "lt" + LTE = "lte" + EQ = "eq" + + +class AlertRule(Base): + __tablename__ = "alert_rules" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + project_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + name: Mapped[str] = mapped_column(String(128), nullable=False) + level: Mapped[AlertLevel] = mapped_column( + SAEnum(AlertLevel), default=AlertLevel.WARNING + ) + field: Mapped[str] = mapped_column( + String(64), nullable=False, default="level" + ) # log field to evaluate + operator: Mapped[AlertOperator] = mapped_column( + SAEnum(AlertOperator), default=AlertOperator.GTE + ) + threshold: Mapped[float] = mapped_column(Float, nullable=False) + window_minutes: Mapped[int] = mapped_column(Integer, default=5) + is_enabled: Mapped[bool] = mapped_column(Boolean, default=True) + notify_channels: Mapped[str] = mapped_column( + String(256), default="" + ) # comma-separated + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + +class AlertHistory(Base): + __tablename__ = "alert_history" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + rule_id: Mapped[str] = mapped_column( + String(36), nullable=False, index=True + ) + project_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + level: Mapped[AlertLevel] = mapped_column(SAEnum(AlertLevel)) + message: Mapped[str] = mapped_column(Text, nullable=False) + triggered_value: Mapped[float] = mapped_column(Float, nullable=True) + is_acknowledged: Mapped[bool] = mapped_column(Boolean, default=False) + triggered_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) + ) diff --git a/backend/app/models/log.py b/backend/app/models/log.py new file mode 100644 index 0000000..683bb46 --- /dev/null +++ b/backend/app/models/log.py @@ -0,0 +1,37 @@ +"""Log entry model — stored in PostgreSQL.""" + +import uuid +from datetime import datetime, timezone + +from sqlalchemy import DateTime, Index, Integer, String, Text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class LogEntry(Base): + __tablename__ = "log_entries" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + project_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + project_name: Mapped[str] = mapped_column(String(128), nullable=False, index=True) + timestamp: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True, + default=lambda: datetime.now(timezone.utc), + ) + level: Mapped[str] = mapped_column(String(16), nullable=False, index=True) + logger: Mapped[str] = mapped_column(String(128), default="root") + message: Mapped[str] = mapped_column(Text, nullable=False) + module: Mapped[str | None] = mapped_column(String(256), nullable=True) + function: Mapped[str | None] = mapped_column(String(128), nullable=True) + line_no: Mapped[int | None] = mapped_column(Integer, nullable=True) + trace_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + exception: Mapped[str | None] = mapped_column(Text, nullable=True) + extra: Mapped[dict] = mapped_column(JSONB, default=dict) + + __table_args__ = ( + Index("ix_log_entries_project_time", "project_id", timestamp.desc()), + ) diff --git a/backend/app/models/project.py b/backend/app/models/project.py new file mode 100644 index 0000000..8a41126 --- /dev/null +++ b/backend/app/models/project.py @@ -0,0 +1,34 @@ +"""Project model — represents an external service that pushes logs.""" + +import uuid +from datetime import datetime, timezone + +from sqlalchemy import String, Boolean, DateTime, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class Project(Base): + __tablename__ = "projects" + + id: Mapped[str] = mapped_column( + String(36), primary_key=True, default=lambda: str(uuid.uuid4()) + ) + name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False) + description: Mapped[str] = mapped_column(Text, default="") + api_key: Mapped[str] = mapped_column( + String(64), unique=True, nullable=False, default=lambda: uuid.uuid4().hex + ) + is_active: Mapped[bool] = mapped_column(Boolean, default=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) + + def __repr__(self) -> str: + return f"" diff --git a/backend/app/schemas/__init__.py b/backend/app/schemas/__init__.py new file mode 100644 index 0000000..8c86a24 --- /dev/null +++ b/backend/app/schemas/__init__.py @@ -0,0 +1,35 @@ +from app.schemas.log import ( + LogEntry, + LogBatchRequest, + LogQueryParams, + LogSearchResult, + LogStatsResult, +) +from app.schemas.project import ( + ProjectCreate, + ProjectUpdate, + ProjectResponse, + ProjectWithToken, +) +from app.schemas.alert import ( + AlertRuleCreate, + AlertRuleUpdate, + AlertRuleResponse, + AlertHistoryResponse, +) + +__all__ = [ + "LogEntry", + "LogBatchRequest", + "LogQueryParams", + "LogSearchResult", + "LogStatsResult", + "ProjectCreate", + "ProjectUpdate", + "ProjectResponse", + "ProjectWithToken", + "AlertRuleCreate", + "AlertRuleUpdate", + "AlertRuleResponse", + "AlertHistoryResponse", +] diff --git a/backend/app/schemas/alert.py b/backend/app/schemas/alert.py new file mode 100644 index 0000000..a9e3ee7 --- /dev/null +++ b/backend/app/schemas/alert.py @@ -0,0 +1,63 @@ +"""Alert schemas.""" + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + + +# ── Alert Rule ───────────────────────────────────────────────── + + +class AlertRuleCreate(BaseModel): + project_id: str = Field(..., min_length=1, max_length=36) + name: str = Field(..., min_length=1, max_length=128) + level: str = Field(default="warning", pattern="^(info|warning|critical)$") + field: str = Field(default="level", max_length=64) + operator: str = Field(default="gte", pattern="^(gt|gte|lt|lte|eq)$") + threshold: float = Field(..., ge=0) + window_minutes: int = Field(default=5, ge=1, le=1440) + notify_channels: List[str] = Field(default_factory=list) + + +class AlertRuleUpdate(BaseModel): + name: Optional[str] = Field(default=None, max_length=128) + level: Optional[str] = Field(default=None, pattern="^(info|warning|critical)$") + field: Optional[str] = Field(default=None, max_length=64) + operator: Optional[str] = Field(default=None, pattern="^(gt|gte|lt|lte|eq)$") + threshold: Optional[float] = Field(default=None, ge=0) + window_minutes: Optional[int] = Field(default=None, ge=1, le=1440) + is_enabled: Optional[bool] = None + notify_channels: Optional[List[str]] = None + + +class AlertRuleResponse(BaseModel): + id: str + project_id: str + name: str + level: str + field: str + operator: str + threshold: float + window_minutes: int + is_enabled: bool + notify_channels: List[str] + created_at: datetime + + model_config = {"from_attributes": True} + + +# ── Alert History ────────────────────────────────────────────── + + +class AlertHistoryResponse(BaseModel): + id: str + rule_id: str + project_id: str + level: str + message: str + triggered_value: Optional[float] = None + is_acknowledged: bool + triggered_at: datetime + + model_config = {"from_attributes": True} diff --git a/backend/app/schemas/log.py b/backend/app/schemas/log.py new file mode 100644 index 0000000..584b038 --- /dev/null +++ b/backend/app/schemas/log.py @@ -0,0 +1,64 @@ +"""Log entry schemas.""" + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + + +class LogEntry(BaseModel): + """A single log entry pushed by a project.""" + + timestamp: Optional[datetime] = None + level: str = Field(default="info", pattern="^(debug|info|warning|error|critical)$") + logger: str = Field(default="root", max_length=128) + message: str = Field(..., min_length=1, max_length=65536) + module: Optional[str] = Field(default=None, max_length=256) + function: Optional[str] = Field(default=None, max_length=128) + line_no: Optional[int] = None + trace_id: Optional[str] = Field(default=None, max_length=64) + exception: Optional[str] = None + extra: Dict[str, Any] = Field(default_factory=dict) + + +class LogBatchRequest(BaseModel): + """Batch of log entries pushed in a single request.""" + + project: str = Field(..., min_length=1, max_length=128) + entries: List[LogEntry] = Field(..., min_length=1, max_length=5000) + + +class LogQueryParams(BaseModel): + """Parameters for log search / query.""" + + project: Optional[str] = None + level: Optional[str] = Field( + default=None, pattern="^(debug|info|warning|error|critical)$" + ) + query: Optional[str] = Field(default=None, max_length=1024) + trace_id: Optional[str] = Field(default=None, max_length=64) + start_time: Optional[datetime] = None + end_time: Optional[datetime] = None + page: int = Field(default=1, ge=1) + page_size: int = Field(default=50, ge=1, le=1000) + sort_by: str = Field(default="timestamp", max_length=32) + sort_order: str = Field(default="desc", pattern="^(asc|desc)$") + + +class LogSearchResult(BaseModel): + """Search result wrapper.""" + + total: int + page: int + page_size: int + hits: List[Dict[str, Any]] + + +class LogStatsResult(BaseModel): + """Aggregated log statistics.""" + + project: Optional[str] = None + time_buckets: List[Dict[str, Any]] + level_counts: Dict[str, int] + top_loggers: List[Dict[str, Any]] + total: int diff --git a/backend/app/schemas/project.py b/backend/app/schemas/project.py new file mode 100644 index 0000000..64e7ba0 --- /dev/null +++ b/backend/app/schemas/project.py @@ -0,0 +1,34 @@ +"""Project schemas.""" + +from datetime import datetime +from typing import Optional + +from pydantic import BaseModel, Field + + +class ProjectCreate(BaseModel): + name: str = Field(..., min_length=1, max_length=128, description="Project name") + description: str = Field(default="", max_length=1024) + + +class ProjectUpdate(BaseModel): + name: Optional[str] = Field(default=None, max_length=128) + description: Optional[str] = Field(default=None, max_length=1024) + is_active: Optional[bool] = None + + +class ProjectResponse(BaseModel): + id: str + name: str + description: str + is_active: bool + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class ProjectWithToken(ProjectResponse): + api_key: str + + model_config = {"from_attributes": True} diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/services/alert_service.py b/backend/app/services/alert_service.py new file mode 100644 index 0000000..e957dc9 --- /dev/null +++ b/backend/app/services/alert_service.py @@ -0,0 +1,142 @@ +"""Alert evaluation and notification service.""" + +import logging +from datetime import datetime, timezone, timedelta +from typing import Optional + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.models.alert import AlertRule, AlertHistory, AlertLevel, AlertOperator +from app.models.log import LogEntry + +logger = logging.getLogger(__name__) + + +class AlertService: + """Evaluates alert rules against recent log data and triggers notifications.""" + + def __init__(self, db: AsyncSession): + self.db = db + + async def evaluate_rules(self) -> list[AlertHistory]: + """Check all enabled alert rules and trigger any that match.""" + result = await self.db.execute( + select(AlertRule).where(AlertRule.is_enabled == True) + ) + rules: list[AlertRule] = result.scalars().all() + triggered: list[AlertHistory] = [] + + for rule in rules: + try: + hit = await self._evaluate_single(rule) + if hit: + triggered.append(hit) + except Exception as e: + logger.error("Error evaluating rule %s: %s", rule.id, e) + + return triggered + + async def _evaluate_single(self, rule: AlertRule) -> Optional[AlertHistory]: + """Evaluate a single rule. Returns AlertHistory if triggered.""" + now = datetime.now(timezone.utc) + since = now - timedelta(minutes=rule.window_minutes) + + stmt = ( + select(func.count(LogEntry.id)) + .where(LogEntry.project_id == rule.project_id) + .where(LogEntry.timestamp >= since) + .where(LogEntry.level == rule.level.value) + ) + count = (await self.db.execute(stmt)).scalar() or 0 + + operator_map = { + AlertOperator.GT: lambda v, t: v > t, + AlertOperator.GTE: lambda v, t: v >= t, + AlertOperator.LT: lambda v, t: v < t, + AlertOperator.LTE: lambda v, t: v <= t, + AlertOperator.EQ: lambda v, t: v == t, + } + + op_fn = operator_map.get(rule.operator) + if op_fn and op_fn(float(count), rule.threshold): + history = AlertHistory( + rule_id=rule.id, + project_id=rule.project_id, + level=rule.level, + message=( + f"Alert '{rule.name}': {rule.level.value} logs count ({count}) " + f"{rule.operator.value} threshold ({rule.threshold}) " + f"in last {rule.window_minutes} min" + ), + triggered_value=float(count), + ) + self.db.add(history) + await self.db.flush() + logger.info("Alert triggered: %s", history.message) + + if rule.notify_channels: + await self._notify(rule, history) + + return history + + return None + + async def _notify(self, rule: AlertRule, history: AlertHistory) -> None: + """Send notification via configured channels.""" + channels = [c.strip() for c in rule.notify_channels.split(",") if c.strip()] + for channel in channels: + if channel == "feishu" and settings.FEISHU_WEBHOOK_URL: + await self._send_feishu(rule, history) + elif channel == "dingtalk" and settings.DINGTALK_WEBHOOK_URL: + await self._send_dingtalk(rule, history) + + async def _send_feishu(self, rule: AlertRule, history: AlertHistory) -> None: + """Send a notification to Feishu webhook.""" + import httpx + + payload = { + "msg_type": "post", + "content": { + "post": { + "zh_cn": { + "title": f"[LogHive] Alert: {rule.name}", + "content": [ + [{"tag": "text", "text": history.message}], + [ + { + "tag": "text", + "text": f"Project: {rule.project_id} | Level: {rule.level.value}", + } + ], + ], + } + } + }, + } + try: + async with httpx.AsyncClient() as client: + await client.post(settings.FEISHU_WEBHOOK_URL, json=payload, timeout=10) + except Exception as e: + logger.error("Feishu notification failed: %s", e) + + async def _send_dingtalk(self, rule: AlertRule, history: AlertHistory) -> None: + """Send a notification to DingTalk webhook.""" + import httpx + + payload = { + "msgtype": "text", + "text": { + "content": ( + f"[LogHive] Alert: {rule.name}\n" + f"{history.message}\n" + f"Project: {rule.project_id} | Level: {rule.level.value}" + ) + }, + } + try: + async with httpx.AsyncClient() as client: + await client.post(settings.DINGTALK_WEBHOOK_URL, json=payload, timeout=10) + except Exception as e: + logger.error("DingTalk notification failed: %s", e) diff --git a/backend/app/services/analytics_service.py b/backend/app/services/analytics_service.py new file mode 100644 index 0000000..dfabd00 --- /dev/null +++ b/backend/app/services/analytics_service.py @@ -0,0 +1,125 @@ +"""Log analytics service — provides aggregated statistics and insights.""" + +import logging +from datetime import datetime, timezone, timedelta +from typing import Optional + +from sqlalchemy import func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import to_display_time +from app.models.log import LogEntry + +logger = logging.getLogger(__name__) + + +class AnalyticsService: + """Time-series analytics over log data.""" + + def __init__(self, db: AsyncSession): + self.db = db + + def _build_filters(self, project: Optional[str], since: datetime): + """Build SQLAlchemy filter list.""" + filters = [LogEntry.timestamp >= since] + if project: + filters.append(LogEntry.project_name == project) + return filters + + async def get_dashboard_stats( + self, project: Optional[str] = None, hours: int = 24 + ) -> dict: + """Get overview statistics for the dashboard.""" + now = datetime.now(timezone.utc) + since = now - timedelta(hours=hours) + filters = self._build_filters(project, since) + + # Total log count + total = (await self.db.execute( + select(func.count(LogEntry.id)).where(*filters) + )).scalar() or 0 + + # Error count + error_count = (await self.db.execute( + select(func.count(LogEntry.id)).where( + *filters, LogEntry.level.in_(["error", "critical"]) + ) + )).scalar() or 0 + + # Level breakdown + level_rows = (await self.db.execute( + select(LogEntry.level, func.count(LogEntry.id)) + .where(*filters) + .group_by(LogEntry.level) + )).all() + level_breakdown = {row[0]: row[1] for row in level_rows} + + # Error rate + error_rate = round((error_count / total * 100), 2) if total > 0 else 0 + + # Top error messages + top_rows = (await self.db.execute( + select(LogEntry.message, func.count(LogEntry.id).label("cnt")) + .where(*filters, LogEntry.level.in_(["error", "critical"])) + .group_by(LogEntry.message) + .order_by(func.count(LogEntry.id).desc()) + .limit(10) + )).all() + top_errors = [{"message": row[0], "count": row[1]} for row in top_rows] + + return { + "total_logs": total, + "error_count": error_count, + "error_rate": error_rate, + "level_breakdown": level_breakdown, + "time_range_hours": hours, + "top_errors": top_errors, + } + + async def get_time_series( + self, + project: Optional[str] = None, + hours: int = 24, + interval: str = "5m", + ) -> list[dict]: + """Get time-series data for log volume charts.""" + now = datetime.now(timezone.utc) + since = now - timedelta(hours=hours) + + n = int("".join(c for c in interval if c.isdigit()) or "5") + unit = "".join(c for c in interval if c.isalpha()) or "m" + if unit == "h": + n = n * 60 + n = max(n, 1) + + # Build WHERE clauses + where_clauses = [f"timestamp >= '{since.isoformat()}'"] + params: dict = {} + if project: + where_clauses.append("project_name = :project") + params["project"] = project + where_sql = " AND ".join(where_clauses) + + sql = text( + f"SELECT " + f" date_trunc('hour', timestamp) + " + f" ((floor(extract(minute from timestamp) / {n})::int * {n}) || ' min')::interval AS bucket, " + f" level, " + f" count(*) AS cnt " + f"FROM log_entries " + f"WHERE {where_sql} " + f"GROUP BY bucket, level " + f"ORDER BY bucket" + ) + + rows = (await self.db.execute(sql, params)).all() + + buckets: dict[str, dict] = {} + for row in rows: + ts = to_display_time(row.bucket) if isinstance(row.bucket, datetime) else str(row.bucket) + if ts not in buckets: + buckets[ts] = {"timestamp": ts, "total": 0, "levels": {}} + buckets[ts]["total"] += row.cnt + buckets[ts]["levels"][row.level] = row.cnt + + return list(buckets.values()) diff --git a/backend/celery_worker/__init__.py b/backend/celery_worker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/celery_worker/celery_app.py b/backend/celery_worker/celery_app.py new file mode 100644 index 0000000..c32f989 --- /dev/null +++ b/backend/celery_worker/celery_app.py @@ -0,0 +1,33 @@ +"""Celery application for async background tasks.""" + +from celery import Celery + +from app.config import settings + +celery_app = Celery( + "loghive", + broker=settings.CELERY_BROKER_URL, + backend=settings.CELERY_RESULT_BACKEND, + include=["celery_worker.tasks"], +) + +celery_app.conf.update( + task_serializer="json", + accept_content=["json"], + result_serializer="json", + timezone="UTC", + enable_utc=True, + task_track_started=True, + task_acks_late=True, + worker_prefetch_multiplier=1, + beat_schedule={ + "evaluate-alert-rules": { + "task": "celery_worker.tasks.evaluate_alerts", + "schedule": settings.ALERT_CHECK_INTERVAL_SECONDS, + }, + "cleanup-old-log-indices": { + "task": "celery_worker.tasks.cleanup_old_indices", + "schedule": 3600, # once per hour + }, + }, +) diff --git a/backend/celery_worker/tasks.py b/backend/celery_worker/tasks.py new file mode 100644 index 0000000..20373c1 --- /dev/null +++ b/backend/celery_worker/tasks.py @@ -0,0 +1,82 @@ +"""Celery tasks for periodic and async processing.""" + +import asyncio +import logging +from datetime import datetime, timezone, timedelta + +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker + +from app.config import settings +from app.models.log import LogEntry +from app.services.alert_service import AlertService + +logger = logging.getLogger(__name__) + +from celery_worker.celery_app import celery_app + + +@celery_app.task(bind=True, max_retries=3) +def evaluate_alerts(self): + """Evaluate all alert rules and trigger notifications. + + Runs periodically (default: every 60 seconds via beat schedule). + """ + try: + loop = asyncio.get_event_loop() + if loop.is_closed(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + result = loop.run_until_complete(_async_evaluate_alerts()) + logger.info("Alert evaluation complete: %d triggers", len(result)) + return {"triggered": len(result)} + except Exception as e: + logger.error("Alert evaluation failed: %s", e) + raise self.retry(exc=e, countdown=30) + + +async def _async_evaluate_alerts(): + """Async body of alert evaluation.""" + engine = create_async_engine(settings.DATABASE_URL) + factory = async_sessionmaker(engine, expire_on_commit=False) + + try: + async with factory() as db: + service = AlertService(db) + triggered = await service.evaluate_rules() + await db.commit() + return triggered + finally: + await engine.dispose() + + +@celery_app.task +def cleanup_old_indices(): + """Delete log entries older than the retention period.""" + try: + loop = asyncio.get_event_loop() + if loop.is_closed(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + result = loop.run_until_complete(_async_cleanup()) + logger.info("Cleanup complete: %d entries deleted", result) + return {"deleted": result} + except Exception as e: + logger.error("Cleanup failed: %s", e) + return {"error": str(e)} + + +async def _async_cleanup() -> int: + """Delete old log entries.""" + engine = create_async_engine(settings.DATABASE_URL) + factory = async_sessionmaker(engine, expire_on_commit=False) + + try: + cutoff = datetime.now(timezone.utc) - timedelta(days=settings.LOG_RETENTION_DAYS) + async with factory() as db: + result = await db.execute(delete(LogEntry).where(LogEntry.timestamp < cutoff)) + await db.commit() + return result.rowcount + finally: + await engine.dispose() diff --git a/backend/requirements.txt b/backend/requirements.txt new file mode 100644 index 0000000..9ec61f1 --- /dev/null +++ b/backend/requirements.txt @@ -0,0 +1,25 @@ +# LogHive Backend Dependencies + +# Framework +fastapi>=0.111.0 +uvicorn[standard]>=0.30.0 +pydantic>=2.7.0 +pydantic-settings>=2.3.0 + +# Database +sqlalchemy[asyncio]>=2.0.30 +asyncpg>=0.29.0 + +# Task Queue +celery>=5.4.0 +redis>=5.0.0 + +# HTTP Client (for webhook notifications) +httpx>=0.27.0 + +# CORS / Middleware +python-multipart>=0.0.9 + +# Utilities +python-dateutil>=2.9.0 +orjson>=3.10.0 diff --git a/client/README.md b/client/README.md new file mode 100644 index 0000000..6e774fc --- /dev/null +++ b/client/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/client/loghive_client/__init__.py b/client/loghive_client/__init__.py new file mode 100644 index 0000000..af3e424 --- /dev/null +++ b/client/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/client/loghive_client/async_client.py b/client/loghive_client/async_client.py new file mode 100644 index 0000000..22475f8 --- /dev/null +++ b/client/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/client/loghive_client/client.py b/client/loghive_client/client.py new file mode 100644 index 0000000..89937c8 --- /dev/null +++ b/client/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/client/loghive_client/handler.py b/client/loghive_client/handler.py new file mode 100644 index 0000000..d45b225 --- /dev/null +++ b/client/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/client/setup.py b/client/setup.py new file mode 100644 index 0000000..8fbf159 --- /dev/null +++ b/client/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", + ], +) diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..f64e327 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,42 @@ +services: + # ── LogHive Backend API ─────────────────────────────────── + backend: + build: + context: ./backend + dockerfile: Dockerfile + ports: + - "8000:8000" + env_file: + - .env + + # ── Celery Worker ───────────────────────────────────────── + celery_worker: + build: + context: ./backend + dockerfile: Dockerfile + command: celery -A celery_worker.celery_app worker --loglevel=info + env_file: + - .env + depends_on: + - backend + + # ── Celery Beat (periodic tasks) ────────────────────────── + celery_beat: + build: + context: ./backend + dockerfile: Dockerfile + command: celery -A celery_worker.celery_app beat --loglevel=info + env_file: + - .env + depends_on: + - backend + + # ── LogHive Frontend (Vue 3) ────────────────────────────── + frontend: + build: + context: ./frontend + dockerfile: Dockerfile + ports: + - "3000:80" + depends_on: + - backend diff --git a/frontend/Dockerfile b/frontend/Dockerfile new file mode 100644 index 0000000..b46ec7d --- /dev/null +++ b/frontend/Dockerfile @@ -0,0 +1,20 @@ +# ── Build Stage ── +FROM node:20-alpine AS builder + +WORKDIR /app + +COPY package.json ./ +RUN npm install + +COPY . . +RUN npm run build + +# ── Production Stage (nginx) ── +FROM nginx:stable-alpine + +COPY --from=builder /app/dist /usr/share/nginx/html +COPY nginx.conf /etc/nginx/conf.d/default.conf + +EXPOSE 80 + +CMD ["nginx", "-g", "daemon off;"] diff --git a/frontend/index.html b/frontend/index.html new file mode 100644 index 0000000..6f6c199 --- /dev/null +++ b/frontend/index.html @@ -0,0 +1,13 @@ + + + + + + LogHive — 日志管理系统 + + + +
+ + + diff --git a/frontend/nginx.conf b/frontend/nginx.conf new file mode 100644 index 0000000..9e45d3b --- /dev/null +++ b/frontend/nginx.conf @@ -0,0 +1,21 @@ +server { + listen 80; + server_name localhost; + + root /usr/share/nginx/html; + index index.html; + + # Vue SPA — serve index.html for all routes + location / { + try_files $uri $uri/ /index.html; + } + + # Proxy API requests to the backend + location /api/ { + proxy_pass http://backend:8000; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + } +} diff --git a/frontend/package.json b/frontend/package.json new file mode 100644 index 0000000..e75641d --- /dev/null +++ b/frontend/package.json @@ -0,0 +1,25 @@ +{ + "name": "loghive-frontend", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "dev": "vite", + "build": "vite build", + "preview": "vite preview" + }, + "dependencies": { + "vue": "^3.4.0", + "vue-router": "^4.3.0", + "pinia": "^2.1.0", + "axios": "^1.7.0", + "echarts": "^5.5.0", + "vue-echarts": "^6.7.0", + "element-plus": "^2.7.0", + "@element-plus/icons-vue": "^2.3.0" + }, + "devDependencies": { + "@vitejs/plugin-vue": "^5.0.0", + "vite": "^5.2.0" + } +} diff --git a/frontend/public/favicon.svg b/frontend/public/favicon.svg new file mode 100644 index 0000000..6337fcc --- /dev/null +++ b/frontend/public/favicon.svg @@ -0,0 +1,4 @@ + + + H + diff --git a/frontend/src/App.vue b/frontend/src/App.vue new file mode 100644 index 0000000..20c1765 --- /dev/null +++ b/frontend/src/App.vue @@ -0,0 +1,69 @@ + + + + + diff --git a/frontend/src/api/index.js b/frontend/src/api/index.js new file mode 100644 index 0000000..21a8348 --- /dev/null +++ b/frontend/src/api/index.js @@ -0,0 +1,92 @@ +import axios from 'axios' + +const api = axios.create({ + baseURL: '/api', + timeout: 10000, +}) + +// Response interceptor for error handling +api.interceptors.response.use( + (response) => response, + (error) => { + const msg = error.response?.data?.detail || error.message + console.error('API Error:', msg) + return Promise.reject(error) + } +) + +// ── Dashboard ───────────────────────────────────────────────── + +export function fetchDashboardStats(params) { + return api.get('/dashboard/stats', { params }) +} + +export function fetchTimeSeries(params) { + return api.get('/dashboard/timeseries', { params }) +} + +// ── Logs ────────────────────────────────────────────────────── + +export function searchLogs(params) { + return api.get('/logs/search', { params }) +} + +export function fetchLogStats(params) { + return api.get('/logs/stats', { params }) +} + +// ── Projects ────────────────────────────────────────────────── + +export function listProjects() { + return api.get('/projects') +} + +export function getProject(id) { + return api.get(`/projects/${id}`) +} + +export function createProject(data) { + return api.post('/projects', data) +} + +export function updateProject(id, data) { + return api.patch(`/projects/${id}`, data) +} + +export function deleteProject(id) { + return api.delete(`/projects/${id}`) +} + +export function rotateApiKey(id) { + return api.post(`/projects/${id}/rotate-key`) +} + +// ── Alerts ──────────────────────────────────────────────────── + +export function listAlertRules(params) { + return api.get('/alerts/rules', { params }) +} + +export function createAlertRule(data) { + return api.post('/alerts/rules', data) +} + +export function updateAlertRule(id, data) { + return api.patch(`/alerts/rules/${id}`, data) +} + +export function deleteAlertRule(id) { + return api.delete(`/alerts/rules/${id}`) +} + +export function listAlertHistory(params) { + return api.get('/alerts/history', { params }) +} + +export function acknowledgeAlert(id) { + return api.post(`/alerts/history/${id}/acknowledge`) +} + +export function healthCheck() { + return api.get('/health') +} diff --git a/frontend/src/main.js b/frontend/src/main.js new file mode 100644 index 0000000..ce000f4 --- /dev/null +++ b/frontend/src/main.js @@ -0,0 +1,20 @@ +import { createApp } from 'vue' +import { createPinia } from 'pinia' +import ElementPlus from 'element-plus' +import 'element-plus/dist/index.css' +import * as ElementPlusIconsVue from '@element-plus/icons-vue' + +import App from './App.vue' +import router from './router' + +const app = createApp(App) + +// Register all Element Plus icons +for (const [key, component] of Object.entries(ElementPlusIconsVue)) { + app.component(key, component) +} + +app.use(createPinia()) +app.use(router) +app.use(ElementPlus) +app.mount('#app') diff --git a/frontend/src/router/index.js b/frontend/src/router/index.js new file mode 100644 index 0000000..7e81bcc --- /dev/null +++ b/frontend/src/router/index.js @@ -0,0 +1,19 @@ +import { createRouter, createWebHistory } from 'vue-router' +import DashboardView from '../views/DashboardView.vue' +import LogSearchView from '../views/LogSearchView.vue' +import ProjectManageView from '../views/ProjectManageView.vue' +import AlertManageView from '../views/AlertManageView.vue' + +const routes = [ + { path: '/', name: 'dashboard', component: DashboardView }, + { path: '/logs', name: 'logs', component: LogSearchView }, + { path: '/projects', name: 'projects', component: ProjectManageView }, + { path: '/alerts', name: 'alerts', component: AlertManageView }, +] + +const router = createRouter({ + history: createWebHistory(), + routes, +}) + +export default router diff --git a/frontend/src/views/AlertManageView.vue b/frontend/src/views/AlertManageView.vue new file mode 100644 index 0000000..12f985d --- /dev/null +++ b/frontend/src/views/AlertManageView.vue @@ -0,0 +1,272 @@ + + + diff --git a/frontend/src/views/DashboardView.vue b/frontend/src/views/DashboardView.vue new file mode 100644 index 0000000..9f1fe30 --- /dev/null +++ b/frontend/src/views/DashboardView.vue @@ -0,0 +1,204 @@ + + + + + diff --git a/frontend/src/views/LogSearchView.vue b/frontend/src/views/LogSearchView.vue new file mode 100644 index 0000000..6e9d37f --- /dev/null +++ b/frontend/src/views/LogSearchView.vue @@ -0,0 +1,222 @@ + + + + + diff --git a/frontend/src/views/ProjectManageView.vue b/frontend/src/views/ProjectManageView.vue new file mode 100644 index 0000000..a39779f --- /dev/null +++ b/frontend/src/views/ProjectManageView.vue @@ -0,0 +1,179 @@ + + + diff --git a/frontend/vite.config.js b/frontend/vite.config.js new file mode 100644 index 0000000..00aec21 --- /dev/null +++ b/frontend/vite.config.js @@ -0,0 +1,15 @@ +import { defineConfig } from 'vite' +import vue from '@vitejs/plugin-vue' + +export default defineConfig({ + plugins: [vue()], + server: { + port: 3000, + proxy: { + '/api': { + target: 'http://localhost:8000', + changeOrigin: true, + }, + }, + }, +}) diff --git a/接入LogHive日志系统.md b/接入LogHive日志系统.md new file mode 100644 index 0000000..e4119c2 --- /dev/null +++ b/接入LogHive日志系统.md @@ -0,0 +1,150 @@ +# 接入 LogHive 日志系统指南 + +## 前置准备 + +向管理员提供你的**项目名称**（如 `user-service`），管理员会返回一个 **API Key**。 + +拿到 API Key 后，在你的项目 `.env` 文件中添加： + +```env +# LogHive 日志系统 +LOGHIVE_ENDPOINT=http://10.10.10.14:8000 +LOGHIVE_PROJECT=你的项目名称 +LOGHIVE_API_KEY=管理员给你的API-Key +``` + +> **不要**把 API 地址和 Key 硬编码在代码里，全部从环境变量读取。 + +--- + +## 方式一：零代码改动（推荐） + +适用于已使用 Python 标准 `logging` 模块的项目。只需在入口文件添加 3 行： + +```python +import logging +import os +from loghive_client import LogHiveHandler + +handler = LogHiveHandler( + project=os.environ["LOGHIVE_PROJECT"], + api_key=os.environ["LOGHIVE_API_KEY"], + endpoint=os.environ["LOGHIVE_ENDPOINT"], + level=logging.INFO, # 只发送 INFO 及以上级别 +) +logging.getLogger().addHandler(handler) + +# 现有代码无需任何修改，所有日志自动发送到 LogHive +logging.info("服务启动成功") +logging.error("数据库连接超时", exc_info=True) +``` + +### 仅发送特定 logger 的日志 + +```python +logger = logging.getLogger("myapp.api") +logger.addHandler(handler) +logger.setLevel(logging.WARNING) +``` + +--- + +## 方式二：使用 LogHive SDK + +适合需要更精细控制的场景，或者不想影响全局 logging 配置。 + +### 安装 + +```bash +pip install /path/to/loghive-client +``` + +### 同步项目使用 + +```python +import os +from loghive_client import LogHiveLogger + +logger = LogHiveLogger( + project=os.environ["LOGHIVE_PROJECT"], + api_key=os.environ["LOGHIVE_API_KEY"], + endpoint=os.environ["LOGHIVE_ENDPOINT"], +) + +logger.info("用户登录成功", user_id=42, ip="1.2.3.4") +logger.warning("API 限流触发", rate="90%") +logger.error("支付回调验签失败", trace_id="req-abc-123", exc_info=True) +logger.debug("缓存命中 key=user:42") +logger.critical("磁盘空间不足，服务即将崩溃") +``` + +SDK 在后台线程异步批量发送，**不会阻塞主线程**。程序退出时会自动 flush 剩余日志。 + +### 异步项目使用（FastAPI / aiohttp） + +```python +import os +from loghive_client import AsyncLogHiveLogger + +async def main(): + async with AsyncLogHiveLogger( + project=os.environ["LOGHIVE_PROJECT"], + api_key=os.environ["LOGHIVE_API_KEY"], + endpoint=os.environ["LOGHIVE_ENDPOINT"], + ) as logger: + await logger.info("请求处理完成", path="/api/users", status=200) +``` + +--- + +## 方式三：直接调用 REST API（非 Python 项目） + +```bash +curl -X POST $LOGHIVE_ENDPOINT/api/logs/ingest \ + -H "Authorization: Bearer $LOGHIVE_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "project": "'$LOGHIVE_PROJECT'", + "entries": [ + { + "level": "error", + "message": "服务异常", + "logger": "myapp.module", + "trace_id": "abc-123", + "extra": {"key": "value"} + } + ] + }' +``` + +--- + +## 日志字段说明 + +| 字段 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `level` | string | 是 | `debug` / `info` / `warning` / `error` / `critical` | +| `message` | string | 是 | 日志内容，最长 65536 字符 | +| `logger` | string | 否 | Logger 名称，默认 `root` | +| `trace_id` | string | 否 | 链路追踪 ID，用于关联跨模块日志 | +| `exception` | string | 否 | 异常堆栈，SDK 通过 `exc_info=True` 自动捕获 | +| `extra` | object | 否 | 任意键值对，支持嵌套结构 | + +--- + +## 在 LogHive 前端查看 + +访问 `http://10.10.10.14:3000`，按项目、级别、关键词、时间范围搜索和统计。 + +--- + +## 常见问题 + +**Q: 发送失败会影响我的主业务吗？** +A: 不会。SDK 在后台线程异步发送，网络失败会自动重试 3 次，最终丢弃并记录本地 warning。 + +**Q: 日志量很大怎么办？** +A: SDK 默认每 2 秒或积攒 50 条批量发送。可通过 `batch_size` 和 `flush_interval` 参数调节。 + +**Q: 多个进程/worker 同时发送有问题吗？** +A: 没问题。每个进程创建自己的 LogHiveLogger 实例即可。