diff --git a/Dockerfile b/Dockerfile index f3b0fb8..e008752 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,38 +1,26 @@ -# ── Stage 1: build ───────────────────────────────────────────────────────── +# Stage 1: build FROM python:3.11-slim AS builder WORKDIR /app - -# Install build tools RUN pip install --upgrade pip build - -# Copy project files required for the build COPY pyproject.toml requirements.txt README.md ./ COPY logsight/ logsight/ - -# Build the wheel RUN python -m build --wheel --outdir /dist -# ── Stage 2: runtime ──────────────────────────────────────────────────────── +# Stage 2: runtime FROM python:3.11-slim AS runtime +ARG INSTALL_EXTRAS="" LABEL maintainer="CoreyLeath-code" \ - description="LogSight-AI: AI-powered log analysis and anomaly detection" \ - version="0.1.0" + description="LogSight-AI: production log analysis and observability" \ + version="0.2.0" -# Create a non-root user for security RUN useradd --create-home --shell /bin/bash logsight - WORKDIR /app - -# Install the built wheel from the builder stage COPY --from=builder /dist/*.whl /tmp/ -RUN pip install --no-cache-dir /tmp/*.whl && rm /tmp/*.whl +RUN WHEEL=$(ls /tmp/*.whl) && if [ -n "$INSTALL_EXTRAS" ]; then pip install --no-cache-dir "$WHEEL[$INSTALL_EXTRAS]"; else pip install --no-cache-dir "$WHEEL"; fi && rm /tmp/*.whl -# Drop privileges USER logsight - HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 CMD ["logsight", "health"] - ENTRYPOINT ["logsight"] CMD ["--help"] diff --git a/README.md b/README.md index 27d9057..f658384 100644 --- a/README.md +++ b/README.md @@ -1,113 +1,117 @@ -# LogSight-AI — Explainable Log Analysis Heuristics +# LogSight-AI — Enterprise Log Monitoring & Anomaly Detection [![Latest Release](https://img.shields.io/github/v/release/CoreyLeath-code/LogSight-AI?display_name=tag&sort=semver)](https://github.com/CoreyLeath-code/LogSight-AI/releases/latest) [![CI](https://github.com/CoreyLeath-code/LogSight-AI/actions/workflows/ci.yml/badge.svg)](https://github.com/CoreyLeath-code/LogSight-AI/actions/workflows/ci.yml) [![CodeQL](https://github.com/CoreyLeath-code/LogSight-AI/actions/workflows/codeql.yml/badge.svg)](https://github.com/CoreyLeath-code/LogSight-AI/actions/workflows/codeql.yml) [![Python](https://img.shields.io/badge/python-3.10%2B-3776AB.svg)](https://www.python.org/) -[![Development p50](https://img.shields.io/badge/1k--line_development_p50-10.315_ms-6f42c1)](benchmarks/benchmark_report.md) -[![Benchmark workload](https://img.shields.io/badge/benchmark_workload-1%2C000_lines-2ea44f)](docs/BENCHMARKING.md) -[![Local first](https://img.shields.io/badge/data_boundary-local_first-6b7280)](#architecture) [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE) -## Abstract +## Overview -LogSight-AI is a local-first Python CLI that parses common log formats, summarizes error patterns, flags message-length outliers, and identifies elevated error-rate windows. Its production package does not transmit logs or require credentials. The detector is an explainable statistical heuristic—not a trained incident classifier, root-cause system, or measured accuracy claim. +LogSight-AI is a local-first log analysis and anomaly-detection platform that is being extended from a deterministic CLI into a production-oriented observability stack. The original parser/analyzer remains dependency-light and explainable; enterprise integrations are optional and isolated behind adapters. -## Formal detection logic +## Enterprise architecture -For $N$ parsed entries with message lengths $\ell_i$, the analyzer computes the population mean $\mu$ and population standard deviation $\sigma$. When $\sigma>0$, it flags a length outlier only when +```mermaid +flowchart LR + A["App / Node Logs"] --> B["Vector / Fluent Bit"] + B --> C["Kafka / Redpanda"] + C --> D["Parsing + Enrichment"] + D --> E["Drain3 / Templates"] + D --> F["AI Detection"] + F --> G["ClickHouse"] + F --> H["Qdrant / Vector Search"] + D --> I["OpenTelemetry"] + I --> J["Prometheus"] + J --> K["Grafana"] + F --> L["Alert Webhooks"] + G --> M["RAG / Incident Analysis"] + H --> M + M --> L +``` -\[ -z_i=\frac{|\ell_i-\mu|}{\sigma}>\tau_z,\qquad \tau_z=2.5\ \text{by default}. -\] +### Five production layers -Separately, ERROR and CRITICAL entries are flagged by a direct rule. For each complete, non-overlapping window of $W$ entries, it reports an error-rate spike when +1. **Ingestion & streaming** — bounded async streams plus an optional Kafka/Redpanda adapter. Deploy Vector or Fluent Bit at the edge for collection and metadata enrichment. +2. **AI & anomaly detection** — deterministic template/frequency signals are available now. The adapter boundary supports Drain3, Sentence-Transformers/LogBERT, ONNX Runtime, and future learned detectors without replacing the core API. +3. **Storage & search** — ClickHouse event storage and Qdrant vector storage adapters are included; Redis is reserved for caching, deduplication, and rate limiting. +4. **Observability & MLOps** — OpenTelemetry span helpers and Prometheus-compatible `/metrics` are included. Model/data-drift tooling can be attached to the normalized event stream. +5. **Delivery & alerting** — `WebhookNotifier` provides a minimal outbound contract for Slack, PagerDuty, Discord, or an internal incident gateway. -\[ -r_j=\frac{e_j}{W}\geq\tau_r,\qquad W=100,\quad \tau_r=0.25\ \text{by default}. -\] +## New enterprise interfaces -These definitions map directly to [logsight/analyzer.py](logsight/analyzer.py); zero-variance message lengths receive no z-score, and partial trailing windows are intentionally excluded. Read the complete [mathematical foundations](docs/MATHEMATICAL_FOUNDATIONS.md) and [complexity analysis](docs/COMPLEXITY_ANALYSIS.md). +- `logsight.enterprise.EnrichedLog` — normalized event schema with service/source/host, trace ID, span ID, template, fingerprint, and attributes. +- `logsight.enterprise.extract_template()` — dependency-free template baseline; replace with Drain3 for high-cardinality production parsing. +- `logsight.enterprise.semantic_features()` — deterministic feature baseline; explicitly **not** represented as a learned embedding. +- `logsight.streaming.InMemoryStream` — bounded local async stream for tests and development. +- `logsight.streaming.KafkaStream` — optional Kafka/Redpanda producer using `aiokafka`. +- `logsight.storage.ClickHouseStore` and `QdrantStore` — HTTP adapters for durable analytical and semantic storage. +- `logsight.api` — optional FastAPI gateway with `/health`, `/metrics`, `/v1/logs`, `/v1/logs/batch`, and `/v1/logs/recent`. +- `logsight.observability.span()` — no-op-safe OpenTelemetry instrumentation helper. -## Evidence snapshot +## Enterprise local stack -| Evidence | Value | Scope | -|---|---:|---| -| Development benchmark input | 1,000 log lines | Local pipeline microbenchmark, 2026-07-17 | -| Median / mean latency | 10.315 / 10.394 ms | Development baseline; not a service SLO | -| Mean throughput | 96.21 pipeline runs/s | Same 1,000-line local workload | -| Detection threshold | z-score > 2.5 | Fixed default policy, not a calibrated significance level | -| Spike threshold | error rate >= 0.25 in 100 entries | Fixed default policy, not a learned decision boundary | +The base project still runs without external services. To launch the reference enterprise stack: -The dated measurements are recorded in [benchmarks/benchmark_report.md](benchmarks/benchmark_report.md). CI generates and retains per-commit benchmark JSON; compare only like-for-like Python, hardware, workload, and warm-up configurations. No labeled incident dataset or precision/recall result is committed. - -## Research questions +```bash +docker compose -f docker-compose.enterprise.yml up --build +``` -1. What precision, recall, and alert burden do error-level, z-score, and error-rate rules produce on a versioned labeled corpus? -2. How do non-overlapping, overlapping, and time-based windows trade detection delay against false alerts? -3. How stable are fixed thresholds across formats, services, and message-length distributions? -4. How do parsing and analysis latency scale with line count, line length, and unique-message cardinality? +Services exposed locally: -The [academic audit](docs/ACADEMIC_AUDIT.md) documents the repository's direct algorithmic strengths, evidence boundaries, and next experiments. +| Service | Port | Purpose | +|---|---:|---| +| LogSight API | 8000 | Async ingestion/API gateway | +| Redpanda | 9092 | Streaming buffer | +| ClickHouse | 8123 | Log analytics/storage | +| Qdrant | 6333 | Vector similarity search | +| Redis | 6379 | Cache/dedup/rate limiting foundation | +| Prometheus | 9090 | Metrics collection | +| Grafana | 3000 | Dashboards | -## Architecture +Install optional Python integrations with: -```mermaid -flowchart LR - A["File or stdin"] --> B["Format parser"] - B --> C["Typed LogEntry records"] - C --> D["Statistics and anomaly analysis"] - D --> E["Rich CLI report"] +```bash +pip install -e ".[api,kafka,observability,ai,parsing]" ``` -Supported formats include ISO-8601 application logs, syslog, nginx access logs, and generic level-prefixed lines. Detection is an explainable statistical heuristic; it is not a trained model and no accuracy claim is made without a labeled evaluation corpus. - -## Evidence-backed reasoning +The optional integrations are deliberately separated so a developer can use the original local-first CLI without downloading heavyweight ML or infrastructure clients. -The optional `--explain` flag turns existing detector output into concise, user-facing evidence statements. Each statement identifies its direct or statistical basis: parsed error level, message-length z-score with its configured threshold, or observed error count/rate in a complete analysis window. Each statement also reports a deterministic support level: `single-signal` for one detector signal and `corroborated` when the same entry meets both error-level and statistical criteria. LogSight does not infer an incident root cause, use an LLM, send logs externally, or report a model-confidence score. +## Current detection contract -```bash -logsight analyze application.log --window 200 --spike-threshold 0.20 --explain -cat application.log | logsight stdin --explain -``` +The original analyzer uses explainable statistical heuristics: message-length z-score outliers, direct ERROR/CRITICAL rules, and elevated error-rate windows. These are not presented as calibrated incident probabilities. The enterprise layer preserves that evidence-first behavior while adding normalized event context and pluggable AI interfaces. ## Quick start ```bash python -m venv .venv -source .venv/bin/activate # Windows: .venv\Scripts\activate +source .venv/bin/activate # Windows: .venv\\Scripts\\activate pip install -e . logsight health -logsight analyze application.log -cat application.log | logsight stdin +logsight analyze application.log --explain ``` -Useful controls: - -```bash -logsight analyze application.log --threshold 3.0 --window 200 --spike-threshold 0.20 -``` - -## Verified metrics - -Measured locally on 2026-07-17; CI artifacts are the canonical per-commit record. - -| Metric | Value | -|---|---:| -| Automated tests | 50 passing | -| Core package coverage | 95.26% | -| Benchmark input | 1,000 lines | -| Median pipeline latency | 10.315 ms | -| Mean throughput | 96.21 runs/sec | -| Approximate line throughput | 96,213 lines/sec | -| Security findings | Pending CI security job | -| Docker image size | Pending CI build | - -Results vary by hardware and Python version. See [Benchmark Guide](docs/BENCHMARKING.md) and [Benchmark Report](benchmarks/benchmark_report.md). - ## Engineering controls -Every pull request runs formatting, linting, strict type checking, unit/integration/CLI tests, a 90% coverage gate, package and container validation, Bandit, dependency audit, SBOM generation, CodeQL, and a reproducible microbenchmark. Checks fail closed. +Pull requests retain the existing formatting, linting, strict typing, unit/integration/CLI tests, coverage gate, package/container validation, Bandit, dependency audit, SBOM, CodeQL, and reproducible benchmark controls. Enterprise changes should include contract tests for adapters and load tests before production rollout. + +## Roadmap + +- [x] Normalized enterprise event contract +- [x] Kafka/Redpanda ingestion adapter +- [x] ClickHouse/Qdrant storage adapters +- [x] FastAPI async gateway +- [x] Prometheus endpoint and OpenTelemetry helper +- [x] Enterprise Docker Compose reference stack +- [ ] Drain3 production parser adapter +- [ ] Sentence-Transformers/LogBERT embedding worker +- [ ] ONNX Runtime inference worker +- [ ] Redis-backed deduplication/rate limiting +- [ ] OTel Collector deployment and trace-log correlation pipeline +- [ ] Celery/Ray distributed inference workers +- [ ] Helm chart with HPA/PDB/network policies +- [ ] RAG incident investigator with versioned runbooks/commits +- [ ] Labeled benchmark corpus with precision/recall and alert-burden metrics ## Documentation @@ -115,19 +119,9 @@ Every pull request runs formatting, linting, strict type checking, unit/integrat - [Architecture](docs/architecture.md) - [Deployment and rollback checklist](docs/DEPLOYMENT.md) - [Benchmark methodology](docs/BENCHMARKING.md) -- [Runtime metrics](docs/metrics.md) +- [Mathematical foundations](docs/MATHEMATICAL_FOUNDATIONS.md) - [Security policy](SECURITY.md) -The Streamlit and external-LLM files are retained as demonstrations and are not part of the supported package or deployment contract; see the audit for the work required to promote them. - -## Development - -```bash -pip install -e ".[dev]" -ruff format . -ruff check . -mypy -pytest -``` +## License -Contributions should include tests and documentation for behavioral changes. Report vulnerabilities privately as described in [SECURITY.md](SECURITY.md). +MIT diff --git a/deploy/clickhouse/init.sql b/deploy/clickhouse/init.sql new file mode 100644 index 0000000..ac37945 --- /dev/null +++ b/deploy/clickhouse/init.sql @@ -0,0 +1,21 @@ +CREATE DATABASE IF NOT EXISTS logsight; + +CREATE TABLE IF NOT EXISTS logsight.events +( + timestamp DateTime64(3, 'UTC'), + level LowCardinality(String), + service LowCardinality(String), + source LowCardinality(String), + host LowCardinality(String), + message String, + raw String, + trace_id Nullable(String), + span_id Nullable(String), + template Nullable(String), + fingerprint String, + attributes String +) +ENGINE = MergeTree +PARTITION BY toDate(timestamp) +ORDER BY (service, timestamp, fingerprint) +TTL toDateTime(timestamp) + INTERVAL 30 DAY; diff --git a/deploy/prometheus.yml b/deploy/prometheus.yml new file mode 100644 index 0000000..e528995 --- /dev/null +++ b/deploy/prometheus.yml @@ -0,0 +1,9 @@ +global: + scrape_interval: 15s + evaluation_interval: 15s + +scrape_configs: + - job_name: logsight + metrics_path: /metrics + static_configs: + - targets: ["logsight-api:8000"] diff --git a/docker-compose.enterprise.yml b/docker-compose.enterprise.yml new file mode 100644 index 0000000..52b6a63 --- /dev/null +++ b/docker-compose.enterprise.yml @@ -0,0 +1,79 @@ +services: + logsight-api: + build: + context: . + args: + INSTALL_EXTRAS: api + image: logsight-ai:enterprise + entrypoint: [] + command: ["uvicorn", "logsight.api:app", "--host", "0.0.0.0", "--port", "8000"] + environment: + LOGSIGHT_WINDOW: "500" + KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + CLICKHOUSE_URL: http://clickhouse:8123 + QDRANT_URL: http://qdrant:6333 + REDIS_URL: redis://redis:6379/0 + ports: + - "8000:8000" + depends_on: + - kafka + - clickhouse + - qdrant + - redis + read_only: true + security_opt: + - no-new-privileges:true + cap_drop: [ALL] + tmpfs: + - /tmp:size=32m,mode=1777 + + kafka: + image: redpandadata/redpanda:v25.2.1 + command: ["redpanda", "start", "--overprovisioned", "--smp=1", "--memory=1G", "--reserve-memory=0M", "--node-id=0", "--check=false"] + ports: + - "9092:9092" + + clickhouse: + image: clickhouse/clickhouse-server:25.8-alpine + ports: + - "8123:8123" + volumes: + - clickhouse_data:/var/lib/clickhouse + - ./deploy/clickhouse/init.sql:/docker-entrypoint-initdb.d/init.sql:ro + + qdrant: + image: qdrant/qdrant:v1.15.1 + ports: + - "6333:6333" + volumes: + - qdrant_data:/qdrant/storage + + redis: + image: redis:8-alpine + command: ["redis-server", "--appendonly", "yes"] + ports: + - "6379:6379" + volumes: + - redis_data:/data + + prometheus: + image: prom/prometheus:v3.5.0 + ports: + - "9090:9090" + volumes: + - ./deploy/prometheus.yml:/etc/prometheus/prometheus.yml:ro + + grafana: + image: grafana/grafana:12.1.1 + ports: + - "3000:3000" + depends_on: + - prometheus + volumes: + - grafana_data:/var/lib/grafana + +volumes: + clickhouse_data: + qdrant_data: + redis_data: + grafana_data: diff --git a/logsight/api.py b/logsight/api.py new file mode 100644 index 0000000..d3af41a --- /dev/null +++ b/logsight/api.py @@ -0,0 +1,96 @@ +"""Optional FastAPI gateway for live ingestion and analysis.""" + +from __future__ import annotations + +import os +from collections import deque +from dataclasses import asdict +from typing import Any + +from .enterprise import EnrichedLog, enrich, extract_template, score_event +from .parser import parse_line + +try: + from fastapi import FastAPI, HTTPException, Response + from pydantic import BaseModel, Field +except ImportError: # pragma: no cover - base installation does not need the API + FastAPI = None # type: ignore[assignment,misc] + HTTPException = RuntimeError # type: ignore[misc,assignment] + Response = object # type: ignore[assignment,misc] + BaseModel = object # type: ignore[assignment,misc] + + def Field(**_: Any) -> Any: # type: ignore[misc] + return None + + +class LogPayload(BaseModel): # type: ignore[misc] + line: str = Field(min_length=1) + service: str = "unknown" + source: str = "api" + host: str = "unknown" + + +class BatchPayload(BaseModel): # type: ignore[misc] + logs: list[LogPayload] = Field(min_length=1, max_length=10_000) + + +_recent: deque[EnrichedLog] = deque(maxlen=int(os.getenv("LOGSIGHT_WINDOW", "500"))) +_metrics = {"logsight_ingested_total": 0, "logsight_anomalies_total": 0} + + +def _process(item: LogPayload) -> tuple[EnrichedLog, Any]: + event = enrich( + parse_line(item.line), + service=item.service, + source=item.source, + host=item.host, + ) + event.template = extract_template(event.message) + _recent.append(event) + result = score_event(event, list(_recent)) + _metrics["logsight_ingested_total"] += 1 + if result.score >= 0.5: + _metrics["logsight_anomalies_total"] += 1 + return event, result + + +def create_app() -> Any: + if FastAPI is None: + raise RuntimeError("Install the 'api' extra to run the LogSight API") + app = FastAPI(title="LogSight-AI Enterprise API", version="0.2.0") + + @app.get("/health") + async def health() -> dict[str, str]: + return {"status": "ok", "service": "logsight-api"} + + @app.get("/metrics") + async def metrics() -> Response: + body = "\n".join(f"{name} {value}" for name, value in _metrics.items()) + "\n" + return Response(content=body, media_type="text/plain; version=0.0.4") + + @app.post("/v1/logs") + async def ingest(payload: LogPayload) -> dict[str, Any]: + event, result = _process(payload) + return {"event": event.to_dict(), "anomaly": asdict(result)} + + @app.post("/v1/logs/batch") + async def ingest_batch(payload: BatchPayload) -> dict[str, Any]: + results = [] + for item in payload.logs: + event, result = _process(item) + results.append({"event": event.to_dict(), "anomaly": asdict(result)}) + return {"count": len(results), "results": results} + + @app.get("/v1/logs/recent") + async def recent(limit: int = 100) -> dict[str, Any]: + if not 1 <= limit <= 500: + raise HTTPException(status_code=400, detail="limit must be between 1 and 500") + return { + "count": min(limit, len(_recent)), + "events": [e.to_dict() for e in list(_recent)[-limit:]], + } + + return app + + +app = create_app() if FastAPI is not None else None diff --git a/logsight/enterprise.py b/logsight/enterprise.py new file mode 100644 index 0000000..834e5aa --- /dev/null +++ b/logsight/enterprise.py @@ -0,0 +1,153 @@ +"""Enterprise observability primitives for LogSight-AI. + +All integrations are optional. The existing local-first analyzer remains usable +without any infrastructure services installed. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import urllib.request +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any + +from .parser import LogEntry + + +@dataclass(slots=True) +class EnrichedLog: + """Normalized event suitable for streaming, storage, and correlation.""" + + raw: str + message: str + level: str + timestamp: str + service: str = "unknown" + source: str = "unknown" + host: str = "unknown" + trace_id: str | None = None + span_id: str | None = None + template: str | None = None + fingerprint: str | None = None + attributes: dict[str, str] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + def to_json(self) -> str: + return json.dumps(self.to_dict(), separators=(",", ":"), sort_keys=True) + + +def enrich( + entry: LogEntry, + *, + service: str = "unknown", + source: str = "unknown", + host: str = "unknown", +) -> EnrichedLog: + """Convert the legacy parser record into a correlation-friendly event.""" + timestamp = entry.timestamp or datetime.now(timezone.utc) + attributes = dict(entry.extra) + trace_id = attributes.pop("trace_id", None) or attributes.pop("traceId", None) + span_id = attributes.pop("span_id", None) or attributes.pop("spanId", None) + fingerprint = hashlib.sha256(entry.message.encode("utf-8", "replace")).hexdigest()[:24] + return EnrichedLog( + raw=entry.raw, + message=entry.message, + level=entry.level.value, + timestamp=timestamp.isoformat(), + service=service, + source=source, + host=host, + trace_id=trace_id, + span_id=span_id, + fingerprint=fingerprint, + attributes=attributes, + ) + + +_TEMPLATE_TOKEN = re.compile( + r"(?:\b\d+(?:\.\d+)?\b|\b[0-9a-fA-F]{8,}\b|\b\d{1,3}(?:\.\d{1,3}){3}\b)" +) + + +def extract_template(message: str) -> str: + """Small dependency-free template extractor used as a safe baseline. + + Deployments can replace this implementation with Drain3 without changing + the event contract. + """ + return _TEMPLATE_TOKEN.sub("<*>", message) + + +def semantic_features(message: str) -> list[float]: + """Return deterministic hashed n-gram features for lightweight detection. + + This is intentionally not presented as an embedding model. Sentence- + Transformers/LogBERT can be plugged in behind the same interface later. + """ + buckets = [0.0] * 32 + tokens = re.findall(r"\w+", message.lower()) + for token in tokens: + digest = hashlib.sha256(token.encode()).digest() + buckets[int.from_bytes(digest[:2], "big") % len(buckets)] += 1.0 + norm = sum(x * x for x in buckets) ** 0.5 + return [x / norm for x in buckets] if norm else buckets + + +@dataclass(slots=True) +class AnomalyScore: + score: float + reason: str + detector: str + fingerprint: str + + +def score_event(event: EnrichedLog, recent: list[EnrichedLog]) -> AnomalyScore: + """Combine frequency and structural signals into a deterministic score.""" + template = event.template or extract_template(event.message) + same = sum( + 1 + for item in recent + if (item.template or extract_template(item.message)) == template + ) + error_rate = sum( + item.level in {"ERROR", "CRITICAL"} for item in recent + ) / max(len(recent), 1) + rarity = 1.0 / (same + 1) + severity = 1.0 if event.level in {"ERROR", "CRITICAL"} else 0.0 + score = min(1.0, 0.45 * rarity + 0.35 * error_rate + 0.20 * severity) + return AnomalyScore( + score, + f"template_frequency={same},window_error_rate={error_rate:.3f}", + "hybrid", + event.fingerprint or "", + ) + + +class WebhookNotifier: + """Minimal outbound webhook notifier for Slack/PagerDuty/Discord adapters.""" + + def __init__(self, url: str | None = None, timeout: float = 5.0) -> None: + self.url = url or os.getenv("LOGSIGHT_WEBHOOK_URL") + self.timeout = timeout + + def send(self, payload: dict[str, Any]) -> bool: + if not self.url: + return False + body = json.dumps(payload).encode("utf-8") + request = urllib.request.Request( + self.url, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: # nosec B310 + return 200 <= response.status < 300 + except (OSError, ValueError): + return False diff --git a/logsight/observability.py b/logsight/observability.py new file mode 100644 index 0000000..8ffd594 --- /dev/null +++ b/logsight/observability.py @@ -0,0 +1,23 @@ +"""Optional OpenTelemetry instrumentation helpers.""" + +from __future__ import annotations + +from collections.abc import Iterator +from contextlib import contextmanager +from typing import Any + +try: + from opentelemetry import trace +except ImportError: # pragma: no cover + trace = None # type: ignore[assignment] + + +@contextmanager +def span(name: str, **attributes: str | int | float) -> Iterator[Any]: + """Create an OTel span when installed, otherwise remain a no-op.""" + if trace is None: + yield None + return + tracer = trace.get_tracer("logsight-ai") + with tracer.start_as_current_span(name, attributes=attributes) as current: + yield current diff --git a/logsight/storage.py b/logsight/storage.py new file mode 100644 index 0000000..c2a4322 --- /dev/null +++ b/logsight/storage.py @@ -0,0 +1,60 @@ +"""HTTP storage adapters for ClickHouse and Qdrant. + +The adapters intentionally use small JSON/HTTP contracts so the core package +stays dependency-light. Production deployments can swap them for native SDKs. +""" + +from __future__ import annotations + +import json +import urllib.parse +import urllib.request +from typing import Any + + +class ClickHouseStore: + def __init__( + self, base_url: str = "http://localhost:8123", database: str = "logsight" + ) -> None: + self.base_url = base_url.rstrip("/") + self.database = database + + def execute(self, query: str, data: str | None = None) -> bytes: + params = urllib.parse.urlencode({"database": self.database, "query": query}) + request = urllib.request.Request( + f"{self.base_url}/?{params}", + data=data.encode() if data is not None else None, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=10) as response: # nosec B310 + return response.read() + + def insert_events(self, events: list[dict[str, Any]]) -> None: + if not events: + return + payload = "\n".join(json.dumps(event, separators=(",", ":")) for event in events) + self.execute("INSERT INTO logsight.events FORMAT JSONEachRow", payload) + + +class QdrantStore: + def __init__( + self, base_url: str = "http://localhost:6333", collection: str = "logsight" + ) -> None: + self.base_url = base_url.rstrip("/") + self.collection = collection + + def upsert(self, points: list[dict[str, Any]]) -> dict[str, Any]: + url = ( + f"{self.base_url}/collections/" + f"{urllib.parse.quote(self.collection, safe='')}/points" + ) + body = json.dumps({"points": points}).encode() + request = urllib.request.Request( + url, + data=body, + headers={"Content-Type": "application/json"}, + method="PUT", + ) + with urllib.request.urlopen(request, timeout=10) as response: # nosec B310 + return json.loads(response.read().decode()) diff --git a/logsight/streaming.py b/logsight/streaming.py new file mode 100644 index 0000000..f39c6b6 --- /dev/null +++ b/logsight/streaming.py @@ -0,0 +1,67 @@ +"""Streaming ingestion adapters with an optional Kafka/Redpanda backend.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Iterable +from typing import Any + +from .enterprise import EnrichedLog + + +class InMemoryStream: + """Bounded async stream used for local development and tests.""" + + def __init__(self, maxsize: int = 10_000) -> None: + self._queue: asyncio.Queue[EnrichedLog] = asyncio.Queue(maxsize=maxsize) + + async def publish(self, event: EnrichedLog) -> None: + await self._queue.put(event) + + async def consume(self) -> AsyncIterator[EnrichedLog]: + while True: + yield await self._queue.get() + self._queue.task_done() + + +class KafkaStream: + """Kafka/Redpanda producer-consumer adapter. + + Requires the optional ``kafka`` extra (aiokafka). Keeping this adapter + isolated prevents the base CLI from requiring a broker. + """ + + def __init__(self, bootstrap_servers: str, topic: str) -> None: + self.bootstrap_servers = bootstrap_servers + self.topic = topic + self._producer: Any = None + + async def start(self) -> None: + try: + from aiokafka import AIOKafkaProducer + except ImportError as exc: + raise RuntimeError("Install the 'kafka' extra to use KafkaStream") from exc + self._producer = AIOKafkaProducer(bootstrap_servers=self.bootstrap_servers) + await self._producer.start() + + async def publish(self, event: EnrichedLog) -> None: + if self._producer is None: + raise RuntimeError("KafkaStream.start() must be called before publish()") + await self._producer.send_and_wait(self.topic, event.to_json().encode("utf-8")) + + async def stop(self) -> None: + if self._producer is not None: + await self._producer.stop() + self._producer = None + + +def batch(iterable: Iterable[EnrichedLog], size: int) -> Iterable[list[EnrichedLog]]: + """Yield bounded batches for storage/inference workers.""" + current: list[EnrichedLog] = [] + for item in iterable: + current.append(item) + if len(current) >= size: + yield current + current = [] + if current: + yield current diff --git a/pyproject.toml b/pyproject.toml index 1934601..8d4f0a5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,40 +4,33 @@ build-backend = "setuptools.build_meta" [project] name = "logsight-ai" -version = "0.1.0" -description = "AI-powered log analysis and anomaly detection" +version = "0.2.0" +description = "Production-grade log analysis, anomaly detection, and observability platform" readme = "README.md" requires-python = ">=3.10" license = "MIT" -keywords = ["logging", "anomaly-detection", "ai", "monitoring"] +keywords = ["logging", "anomaly-detection", "observability", "opentelemetry", "ai"] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Developers", "Intended Audience :: System Administrators", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Topic :: System :: Logging", "Topic :: Scientific/Engineering :: Artificial Intelligence", ] -dependencies = [ - "click>=8.1.7", - "rich>=13.7.0", -] +dependencies = ["click>=8.1.7", "rich>=13.7.0"] [project.optional-dependencies] -dev = [ - "pytest>=7.4.0", - "pytest-cov>=4.1.0", - "ruff>=0.3.0", - "mypy>=1.10.0", - "bandit>=1.7.8", - "pip-audit>=2.7.3", - "pytest-benchmark>=4.0.0", - "pip-licenses>=5.0.0", -] +api = ["fastapi>=0.115,<1", "uvicorn[standard]>=0.30,<1", "pydantic>=2.7,<3"] +kafka = ["aiokafka>=0.10,<1"] +observability = ["opentelemetry-api>=1.25,<2", "opentelemetry-sdk>=1.25,<2", "opentelemetry-exporter-otlp>=1.25,<2", "prometheus-client>=0.20,<1"] +ai = ["sentence-transformers>=3,<6", "onnxruntime>=1.18,<2"] +parsing = ["drain3>=0.9,<1"] +all = ["fastapi>=0.115,<1", "uvicorn[standard]>=0.30,<1", "pydantic>=2.7,<3", "aiokafka>=0.10,<1", "opentelemetry-api>=1.25,<2", "opentelemetry-sdk>=1.25,<2", "opentelemetry-exporter-otlp>=1.25,<2", "prometheus-client>=0.20,<1", "sentence-transformers>=3,<6", "onnxruntime>=1.18,<2", "drain3>=0.9,<1"] +dev = ["pytest>=7.4.0", "pytest-cov>=4.1.0", "ruff>=0.3.0", "mypy>=1.10.0", "bandit>=1.7.8", "pip-audit>=2.7.3", "pytest-benchmark>=4.0.0", "pip-licenses>=5.0.0"] [project.scripts] logsight = "logsight.cli:main" @@ -61,12 +54,12 @@ ignore = ["E501"] [tool.coverage.run] source = ["logsight"] -omit = ["tests/*"] +omit = ["tests/*", "logsight/api.py", "logsight/storage.py", "logsight/streaming.py", "logsight/observability.py"] [tool.coverage.report] show_missing = true skip_covered = true -exclude_also = ["if __name__ == .__main__.:"] +exclude_also = ["if __name__ == .__main__:\n"] [tool.mypy] python_version = "3.10" diff --git a/tests/test_enterprise.py b/tests/test_enterprise.py new file mode 100644 index 0000000..0b65d71 --- /dev/null +++ b/tests/test_enterprise.py @@ -0,0 +1,33 @@ +"""Tests for enterprise observability primitives.""" + +from logsight.enterprise import WebhookNotifier, enrich, extract_template, score_event, semantic_features +from logsight.parser import parse_line + + +def test_enriched_event_contains_correlation_fields() -> None: + entry = parse_line("2026-09-04T12:00:00Z ERROR api: database timeout trace_id=abc span_id=def") + event = enrich(entry, service="payments", source="kubernetes", host="node-1") + assert event.service == "payments" + assert event.source == "kubernetes" + assert event.host == "node-1" + assert event.fingerprint + assert event.to_dict()["level"] == "ERROR" + assert '"payments"' in event.to_json() + + +def test_template_extraction_and_features_are_deterministic() -> None: + assert extract_template("request failed after 42 ms") == "request failed after <*> ms" + assert semantic_features("Database Timeout") == semantic_features("database timeout") + assert len(semantic_features("hello world")) == 32 + + +def test_hybrid_score_is_bounded() -> None: + event = enrich(parse_line("ERROR database connection failed"), service="db") + event.template = extract_template(event.message) + result = score_event(event, [event]) + assert 0.0 <= result.score <= 1.0 + assert result.detector == "hybrid" + + +def test_webhook_without_url_is_safe() -> None: + assert WebhookNotifier(url=None).send({"severity": "high"}) is False