diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ba1e8ba..16d855a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -86,6 +86,9 @@ jobs: lint-python: name: Lint (Python) runs-on: ubuntu-latest + strategy: + matrix: + service: [log-producer, llm-query-python] steps: - uses: actions/checkout@v4 @@ -98,12 +101,31 @@ jobs: run: pip install ruff==0.7.4 - name: Run ruff check - working-directory: services/log-producer + working-directory: services/${{ matrix.service }} run: ruff check . - name: Run ruff format check - working-directory: services/log-producer + working-directory: services/${{ matrix.service }} run: ruff format --check . + test-python: + name: Test (Python - llm-query) + runs-on: ubuntu-latest + needs: lint-python + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install dependencies + working-directory: services/llm-query-python + run: pip install -r requirements-dev.txt + + - name: Run tests + working-directory: services/llm-query-python + run: python -m pytest tests/ -v test-go-integration: name: Integration Test (Go + TimescaleDB) runs-on: ubuntu-latest @@ -140,7 +162,7 @@ jobs: docker-build: name: Docker Build Validation runs-on: ubuntu-latest - needs: [test-go, lint-python] + needs: [test-go, lint-test-rust, test-python] steps: - uses: actions/checkout@v4 @@ -166,11 +188,17 @@ jobs: context: services/anomaly-rust push: false tags: sentinelops/anomaly-rust:ci + - name: Build llm-query image + uses: docker/build-push-action@v6 + with: + context: services/llm-query-python + push: false + tags: sentinelops/llm-query-python:ci ci-success: name: CI Success runs-on: ubuntu-latest needs: - [lint-go, test-go, test-go-integration, lint-test-rust, lint-python, docker-build] + [lint-go, test-go, test-go-integration, lint-test-rust, lint-python, test-python, docker-build] if: always() steps: - name: Check all jobs succeeded diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index 8e2cd08..9199668 100644 --- a/infra/docker-compose.yml +++ b/infra/docker-compose.yml @@ -60,6 +60,7 @@ services: ports: - "6379:6379" networks: [sentinelops-net] + log-producer: build: context: ../services/log-producer @@ -75,6 +76,7 @@ services: ANOMALY_PROBABILITY: "0.02" restart: unless-stopped networks: [sentinelops-net] + ingestion-go: build: context: ../services/ingestion-go @@ -97,6 +99,7 @@ services: - "9100:9100" restart: unless-stopped networks: [sentinelops-net] + anomaly-rust: build: context: ../services/anomaly-rust @@ -119,6 +122,29 @@ services: - "9200:9200" restart: unless-stopped networks: [sentinelops-net] + + llm-query: + build: + context: ../services/llm-query-python + dockerfile: Dockerfile + container_name: sentinelops-llm-query + depends_on: + timescaledb: + condition: service_healthy + environment: + DATABASE_URL: postgres://sentinelops:${TIMESCALE_PASSWORD:-devpassword}@timescaledb:5432/sentinelops + OLLAMA_BASE_URL: http://host.docker.internal:11434 + OLLAMA_MODEL: llama3.2:3b + OLLAMA_KEEP_ALIVE: 10m + OLLAMA_NUM_PREDICT: "300" + OLLAMA_TIMEOUT_SECONDS: "180" + extra_hosts: + - "host.docker.internal:host-gateway" + ports: + - "9300:9300" + restart: unless-stopped + networks: [sentinelops-net] + networks: sentinelops-net: driver: bridge diff --git a/services/llm-query-python/.gitignore b/services/llm-query-python/.gitignore new file mode 100644 index 0000000..b6c3b5e --- /dev/null +++ b/services/llm-query-python/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +.venv/ +*.pyc +.pytest_cache/ diff --git a/services/llm-query-python/Dockerfile b/services/llm-query-python/Dockerfile new file mode 100644 index 0000000..4376520 --- /dev/null +++ b/services/llm-query-python/Dockerfile @@ -0,0 +1,17 @@ +FROM python:3.12-slim + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY app/ ./app/ + +ENV DATABASE_URL=postgresql://sentinelops:devpassword@timescaledb:5432/sentinelops \ + OLLAMA_BASE_URL=http://host.docker.internal:11434 \ + OLLAMA_MODEL=llama3.2:3b \ + METRICS_PORT=9300 + +EXPOSE 9300 + +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "9300"] \ No newline at end of file diff --git a/services/llm-query-python/app/__init__.py b/services/llm-query-python/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/llm-query-python/app/config.py b/services/llm-query-python/app/config.py new file mode 100644 index 0000000..2c91b3c --- /dev/null +++ b/services/llm-query-python/app/config.py @@ -0,0 +1,22 @@ +"""Configuration loaded from environment variables.""" + +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + database_url: str = ( + "postgresql://sentinelops:devpassword@localhost:5432/sentinelops" + ) + ollama_base_url: str = "http://host.docker.internal:11434" + ollama_model: str = "llama3.2:3b" + ollama_keep_alive: str = "10m" + ollama_num_predict: int = 300 + ollama_timeout_seconds: float = 180.0 + default_lookback_hours: int = 6 + metrics_port: int = 9300 + + class Config: + env_prefix = "" + + +settings = Settings() diff --git a/services/llm-query-python/app/jobs.py b/services/llm-query-python/app/jobs.py new file mode 100644 index 0000000..1063d05 --- /dev/null +++ b/services/llm-query-python/app/jobs.py @@ -0,0 +1,82 @@ +"""In-memory async job queue for LLM query requests. + +Ollama inference on constrained hardware can take tens of seconds to +minutes, which makes a blocking HTTP request/response the wrong shape for +this API. Instead, POST /query enqueues a job and returns immediately; +GET /query/{job_id} polls for status and result. + +This is intentionally in-memory (a dict), not Redis-backed, for now: it's +simple, correct for a single-process deployment, and the documented +upgrade path (Redis is already provisioned in the stack) is to swap the +JobStore implementation without changing the API contract, once multi- +worker or multi-process deployment is actually needed. +""" + +import asyncio +import uuid +from datetime import datetime, timezone +from enum import Enum + +from pydantic import BaseModel + + +class JobStatus(str, Enum): + PENDING = "pending" + RUNNING = "running" + COMPLETE = "complete" + FAILED = "failed" + + +class Job(BaseModel): + job_id: str + status: JobStatus + query: str + created_at: datetime + completed_at: datetime | None = None + result: str | None = None + error: str | None = None + + +class JobStore: + def __init__(self) -> None: + self._jobs: dict[str, Job] = {} + self._lock = asyncio.Lock() + + async def create(self, query: str) -> Job: + job = Job( + job_id=str(uuid.uuid4()), + status=JobStatus.PENDING, + query=query, + created_at=datetime.now(timezone.utc), + ) + async with self._lock: + self._jobs[job.job_id] = job + return job + + async def get(self, job_id: str) -> Job | None: + async with self._lock: + return self._jobs.get(job_id) + + async def mark_running(self, job_id: str) -> None: + async with self._lock: + if job_id in self._jobs: + self._jobs[job_id].status = JobStatus.RUNNING + + async def mark_complete(self, job_id: str, result: str) -> None: + async with self._lock: + if job_id in self._jobs: + job = self._jobs[job_id] + job.status = JobStatus.COMPLETE + job.result = result + job.completed_at = datetime.now(timezone.utc) + + async def mark_failed(self, job_id: str, error: str) -> None: + async with self._lock: + if job_id in self._jobs: + job = self._jobs[job_id] + job.status = JobStatus.FAILED + job.error = error + job.completed_at = datetime.now(timezone.utc) + + +job_store = JobStore() diff --git a/services/llm-query-python/app/main.py b/services/llm-query-python/app/main.py new file mode 100644 index 0000000..8527437 --- /dev/null +++ b/services/llm-query-python/app/main.py @@ -0,0 +1,128 @@ +"""SentinelOps LLM Query Service. + +Exposes a natural-language query interface over detected anomalies and +service stats, grounded via retrieval against TimescaleDB (RAG), with +generation via a locally-hosted Ollama model. +""" + +import asyncio +import logging +from contextlib import asynccontextmanager + +import asyncpg +from fastapi import FastAPI, HTTPException +from fastapi.responses import PlainTextResponse +from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST +from pydantic import BaseModel + +from app.config import settings +from app.jobs import Job, JobStatus, job_store +from app.ollama_client import OllamaError, generate +from app.retrieval import format_context_for_prompt, retrieve_context + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s" +) +logger = logging.getLogger("llm-query-service") + +QUERIES_TOTAL = Counter( + "sentinelops_llm_queries_total", "Total number of queries submitted." +) +QUERY_FAILURES_TOTAL = Counter( + "sentinelops_llm_query_failures_total", "Total number of failed queries." +) +QUERY_DURATION = Histogram( + "sentinelops_llm_query_duration_seconds", + "Time from job creation to completion, including LLM generation.", + buckets=[1, 5, 15, 30, 60, 120, 180, 300], +) + +SYSTEM_PROMPT = ( + "You are an SRE assistant analyzing distributed system telemetry. " + "Answer questions using ONLY the context data provided below. " + "If the context doesn't contain enough information to answer, say so " + "explicitly rather than guessing. Be concise and specific — cite actual " + "numbers from the context. Do not invent services, timestamps, or metrics " + "that are not present in the context." +) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + app.state.pool = await asyncpg.create_pool( + settings.database_url, min_size=1, max_size=5 + ) + logger.info("connected to TimescaleDB") + yield + await app.state.pool.close() + + +app = FastAPI(title="SentinelOps LLM Query Service", lifespan=lifespan) + + +class QueryRequest(BaseModel): + query: str + service: str | None = None + lookback_hours: int | None = None + + +class QueryAccepted(BaseModel): + job_id: str + status: JobStatus + + +async def _run_query_job( + job_id: str, request: QueryRequest, pool: asyncpg.Pool +) -> None: + await job_store.mark_running(job_id) + try: + context = await retrieve_context( + pool, + lookback_hours=request.lookback_hours or settings.default_lookback_hours, + service_filter=request.service, + ) + context_text = format_context_for_prompt(context) + + prompt = f"{context_text}\n\nQuestion: {request.query}\n\nAnswer:" + answer = await generate(prompt, system=SYSTEM_PROMPT) + + await job_store.mark_complete(job_id, answer) + except OllamaError as e: + logger.error("ollama generation failed", exc_info=e) + QUERY_FAILURES_TOTAL.inc() + await job_store.mark_failed(job_id, str(e)) + except Exception as e: # noqa: BLE001 - job failures must never crash the worker + logger.error("unexpected error processing query job", exc_info=e) + QUERY_FAILURES_TOTAL.inc() + await job_store.mark_failed(job_id, "internal error processing query") + + +@app.post("/query", response_model=QueryAccepted, status_code=202) +async def submit_query(request: QueryRequest) -> QueryAccepted: + QUERIES_TOTAL.inc() + job = await job_store.create(request.query) + + async def _tracked_run(): + with QUERY_DURATION.time(): + await _run_query_job(job.job_id, request, app.state.pool) + + asyncio.create_task(_tracked_run()) + return QueryAccepted(job_id=job.job_id, status=job.status) + + +@app.get("/query/{job_id}", response_model=Job) +async def get_query_result(job_id: str) -> Job: + job = await job_store.get(job_id) + if job is None: + raise HTTPException(status_code=404, detail="job not found") + return job + + +@app.get("/healthz") +async def healthz() -> dict[str, str]: + return {"status": "ok"} + + +@app.get("/metrics") +async def metrics() -> PlainTextResponse: + return PlainTextResponse(generate_latest(), media_type=CONTENT_TYPE_LATEST) diff --git a/services/llm-query-python/app/ollama_client.py b/services/llm-query-python/app/ollama_client.py new file mode 100644 index 0000000..e23f893 --- /dev/null +++ b/services/llm-query-python/app/ollama_client.py @@ -0,0 +1,40 @@ +"""Thin async client for Ollama's generate API.""" + +import httpx + +from app.config import settings + + +class OllamaError(Exception): + pass + + +async def generate(prompt: str, system: str | None = None) -> str: + """Calls Ollama's /api/generate with streaming disabled. Uses keep_alive + to avoid paying the ~35s model-load penalty on every request, and caps + num_predict so a single query has a bounded worst-case latency rather + than running unbounded on constrained hardware. + """ + payload = { + "model": settings.ollama_model, + "prompt": prompt, + "stream": False, + "keep_alive": settings.ollama_keep_alive, + "options": { + "num_predict": settings.ollama_num_predict, + }, + } + if system: + payload["system"] = system + + try: + async with httpx.AsyncClient(timeout=settings.ollama_timeout_seconds) as client: + response = await client.post( + f"{settings.ollama_base_url}/api/generate", + json=payload, + ) + response.raise_for_status() + data = response.json() + return data.get("response", "").strip() + except httpx.HTTPError as e: + raise OllamaError(f"Ollama request failed: {e}") from e diff --git a/services/llm-query-python/app/retrieval.py b/services/llm-query-python/app/retrieval.py new file mode 100644 index 0000000..2aaacee --- /dev/null +++ b/services/llm-query-python/app/retrieval.py @@ -0,0 +1,147 @@ +"""Retrieval layer: pulls grounded context from TimescaleDB for a query. + +This is the "R" in RAG — we never let the LLM invent incident data. Every +answer is grounded in rows actually read from anomalies / log_events_1min. +""" + +from datetime import datetime, timedelta, timezone + +import asyncpg +from pydantic import BaseModel + + +class AnomalyContext(BaseModel): + detected_at: datetime + bucket_ts: datetime + service: str + anomaly_type: str + metric_value: float + baseline_mean: float + z_score: float + severity: str + + +class ServiceStatsContext(BaseModel): + service: str + total_events: int + total_errors: int + avg_p95_latency_ms: float | None + max_latency_ms: float | None + + +class RetrievedContext(BaseModel): + window_start: datetime + window_end: datetime + anomalies: list[AnomalyContext] + service_stats: list[ServiceStatsContext] + + def is_empty(self) -> bool: + return not self.anomalies and not self.service_stats + + +async def retrieve_context( + pool: asyncpg.Pool, + lookback_hours: int, + service_filter: str | None = None, +) -> RetrievedContext: + """Fetches recent anomalies and per-service aggregate stats within the + lookback window. If service_filter is given, scopes both queries to + that service; otherwise returns a system-wide view. + """ + window_end = datetime.now(timezone.utc) + window_start = window_end - timedelta(hours=lookback_hours) + + anomaly_query = """ + SELECT detected_at, bucket_ts, service, anomaly_type, + metric_value, baseline_mean, z_score, severity + FROM anomalies + WHERE detected_at >= $1 + AND ($2::text IS NULL OR service = $2) + ORDER BY detected_at DESC + LIMIT 50 + """ + stats_query = """ + SELECT service, + sum(event_count) AS total_events, + sum(error_count) AS total_errors, + avg(p95_latency_ms) AS avg_p95_latency_ms, + max(max_latency_ms) AS max_latency_ms + FROM log_events_1min + WHERE bucket >= $1 + AND ($2::text IS NULL OR service = $2) + GROUP BY service + ORDER BY service + """ + + async with pool.acquire() as conn: + anomaly_rows = await conn.fetch(anomaly_query, window_start, service_filter) + stats_rows = await conn.fetch(stats_query, window_start, service_filter) + + anomalies = [ + AnomalyContext( + detected_at=r["detected_at"], + bucket_ts=r["bucket_ts"], + service=r["service"], + anomaly_type=r["anomaly_type"], + metric_value=r["metric_value"], + baseline_mean=r["baseline_mean"], + z_score=r["z_score"], + severity=r["severity"], + ) + for r in anomaly_rows + ] + service_stats = [ + ServiceStatsContext( + service=r["service"], + total_events=r["total_events"] or 0, + total_errors=r["total_errors"] or 0, + avg_p95_latency_ms=r["avg_p95_latency_ms"], + max_latency_ms=r["max_latency_ms"], + ) + for r in stats_rows + ] + + return RetrievedContext( + window_start=window_start, + window_end=window_end, + anomalies=anomalies, + service_stats=service_stats, + ) + + +def format_context_for_prompt(context: RetrievedContext) -> str: + """Renders retrieved context as compact, LLM-readable text. Kept terse + deliberately — every token here costs generation latency on constrained + hardware, so we favor a dense table-like format over prose. + """ + lines = [ + f"Time window: {context.window_start.isoformat()} to {context.window_end.isoformat()}", + "", + ] + + if context.service_stats: + lines.append("Service stats:") + for s in context.service_stats: + error_rate = ( + (s.total_errors / s.total_events * 100) if s.total_events else 0.0 + ) + lines.append( + f"- {s.service}: {s.total_events} events, " + f"{error_rate:.1f}% error rate, " + f"avg p95 latency {s.avg_p95_latency_ms or 0:.0f}ms, " + f"max latency {s.max_latency_ms or 0:.0f}ms" + ) + lines.append("") + + if context.anomalies: + lines.append("Detected anomalies (most recent first):") + for a in context.anomalies: + lines.append( + f"- [{a.severity.upper()}] {a.service} at {a.bucket_ts.isoformat()}: " + f"{a.anomaly_type}, value={a.metric_value:.2f}, " + f"baseline={a.baseline_mean:.2f}, z_score={a.z_score:.2f}" + ) + else: + lines.append("No anomalies detected in this window.") + + return "\n".join(lines) diff --git a/services/llm-query-python/requirements-dev.txt b/services/llm-query-python/requirements-dev.txt new file mode 100644 index 0000000..3cadefe --- /dev/null +++ b/services/llm-query-python/requirements-dev.txt @@ -0,0 +1,4 @@ +-r requirements.txt +pytest==8.3.3 +pytest-asyncio==0.24.0 +ruff==0.7.4 \ No newline at end of file diff --git a/services/llm-query-python/requirements.txt b/services/llm-query-python/requirements.txt new file mode 100644 index 0000000..f829282 --- /dev/null +++ b/services/llm-query-python/requirements.txt @@ -0,0 +1,8 @@ +fastapi==0.115.5 +uvicorn[standard]==0.32.1 +asyncpg==0.30.0 +httpx==0.27.2 +pydantic==2.10.2 +pydantic-settings==2.6.1 +prometheus-client==0.21.1 +python-dotenv==1.0.1 \ No newline at end of file diff --git a/services/llm-query-python/tests/pyproject.toml b/services/llm-query-python/tests/pyproject.toml new file mode 100644 index 0000000..60392ea --- /dev/null +++ b/services/llm-query-python/tests/pyproject.toml @@ -0,0 +1,10 @@ +[tool.ruff] +line-length = 100 +target-version = "py312" + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP", "B", "SIM"] +ignore = ["B008"] # FastAPI's Depends()/default-arg pattern triggers this; it's idiomatic here + +[tool.ruff.format] +quote-style = "double" \ No newline at end of file diff --git a/services/llm-query-python/tests/pytest.ini b/services/llm-query-python/tests/pytest.ini new file mode 100644 index 0000000..ac7d986 --- /dev/null +++ b/services/llm-query-python/tests/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +asyncio_mode = strict +asyncio_default_fixture_loop_scope = function \ No newline at end of file diff --git a/services/llm-query-python/tests/test_jobs.py b/services/llm-query-python/tests/test_jobs.py new file mode 100644 index 0000000..4e45b1c --- /dev/null +++ b/services/llm-query-python/tests/test_jobs.py @@ -0,0 +1,40 @@ +import pytest +from app.jobs import JobStatus, JobStore + + +@pytest.mark.asyncio +async def test_job_lifecycle_success(): + store = JobStore() + job = await store.create("what happened?") + assert job.status == JobStatus.PENDING + assert job.result is None + + await store.mark_running(job.job_id) + running = await store.get(job.job_id) + assert running.status == JobStatus.RUNNING + + await store.mark_complete(job.job_id, "here is the answer") + done = await store.get(job.job_id) + assert done.status == JobStatus.COMPLETE + assert done.result == "here is the answer" + assert done.completed_at is not None + + +@pytest.mark.asyncio +async def test_job_lifecycle_failure(): + store = JobStore() + job = await store.create("bad query") + await store.mark_running(job.job_id) + await store.mark_failed(job.job_id, "ollama unreachable") + + failed = await store.get(job.job_id) + assert failed.status == JobStatus.FAILED + assert failed.error == "ollama unreachable" + assert failed.result is None + + +@pytest.mark.asyncio +async def test_get_unknown_job_returns_none(): + store = JobStore() + result = await store.get("does-not-exist") + assert result is None diff --git a/services/llm-query-python/tests/test_retrieval.py b/services/llm-query-python/tests/test_retrieval.py new file mode 100644 index 0000000..e539ed6 --- /dev/null +++ b/services/llm-query-python/tests/test_retrieval.py @@ -0,0 +1,93 @@ +from datetime import UTC, datetime + +from app.retrieval import ( + AnomalyContext, + RetrievedContext, + ServiceStatsContext, + format_context_for_prompt, +) + + +def make_window(): + return ( + datetime(2026, 7, 22, 6, 0, tzinfo=UTC), + datetime(2026, 7, 22, 12, 0, tzinfo=UTC), + ) + + +def test_format_context_empty_window(): + start, end = make_window() + context = RetrievedContext(window_start=start, window_end=end, anomalies=[], service_stats=[]) + text = format_context_for_prompt(context) + assert "No anomalies detected" in text + assert context.is_empty() + + +def test_format_context_includes_service_stats_and_error_rate(): + start, end = make_window() + context = RetrievedContext( + window_start=start, + window_end=end, + anomalies=[], + service_stats=[ + ServiceStatsContext( + service="checkout", + total_events=200, + total_errors=20, + avg_p95_latency_ms=150.0, + max_latency_ms=300.0, + ) + ], + ) + text = format_context_for_prompt(context) + assert "checkout" in text + # 20/200 = 10.0% error rate — verifies the computed field, not just passthrough + assert "10.0% error rate" in text + assert not context.is_empty() + + +def test_format_context_includes_anomalies_with_severity(): + start, end = make_window() + context = RetrievedContext( + window_start=start, + window_end=end, + anomalies=[ + AnomalyContext( + detected_at=end, + bucket_ts=end, + service="payments", + anomaly_type="error_burst", + metric_value=0.5, + baseline_mean=0.12, + z_score=3.29, + severity="warning", + ) + ], + service_stats=[], + ) + text = format_context_for_prompt(context) + assert "[WARNING]" in text + assert "payments" in text + assert "error_burst" in text + assert "z_score=3.29" in text + + +def test_format_context_handles_zero_events_without_division_error(): + start, end = make_window() + context = RetrievedContext( + window_start=start, + window_end=end, + anomalies=[], + service_stats=[ + ServiceStatsContext( + service="idle-service", + total_events=0, + total_errors=0, + avg_p95_latency_ms=None, + max_latency_ms=None, + ) + ], + ) + # Must not raise ZeroDivisionError + text = format_context_for_prompt(context) + assert "0.0% error rate" in text