diff --git a/src/nsct/api/main.py b/src/nsct/api/main.py index 9e8bc09..913ec10 100644 --- a/src/nsct/api/main.py +++ b/src/nsct/api/main.py @@ -3,17 +3,22 @@ from __future__ import annotations import logging +import os +import time +import uuid from contextlib import asynccontextmanager import httpx -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import Response from nsct.config import AppSettings +from nsct.logging_config import get_logger, set_request_ctx +from nsct.metrics import G_ACTIVE_RESEARCH_RUNS, metrics logger = logging.getLogger(__name__) - # --------------------------------------------------------------------------- # Lifespan # --------------------------------------------------------------------------- @@ -41,6 +46,48 @@ async def lifespan(app: FastAPI): yield + logger.info("NSCT API shutting down") + metrics.gauge(G_ACTIVE_RESEARCH_RUNS, 0) + + +# --------------------------------------------------------------------------- +# Request-logging middleware +# --------------------------------------------------------------------------- + + +async def request_logging_middleware(request: Request, call_next): + """Middleware: log request metadata (request_id, method, path, status_code, duration_ms).""" + request_id = str(uuid.uuid4()) + start_time = time.monotonic() + + # Set request context + set_request_ctx( + request_id=request_id, + method=request.method, + path=request.url.path, + ) + + response = await call_next(request) + + duration_ms = round((time.monotonic() - start_time) * 1000, 2) + status_code = response.status_code + + set_request_ctx(status_code=str(status_code), duration_ms=str(duration_ms)) + + logger.info( + "HTTP %s %s %s %d %.1fms", + request.method, + request.url.path, + request_id, + status_code, + duration_ms, + ) + + # Inject request_id header for client-side tracing + response.headers["X-Request-ID"] = request_id + + return response + # --------------------------------------------------------------------------- # App factory @@ -71,6 +118,9 @@ def create_app() -> FastAPI: allow_headers=["*"], ) + # Request-logging middleware + app.middleware("http")(request_logging_middleware) + # Mount health router from nsct.api.health import router as health_router app.include_router(health_router, tags=["system"]) @@ -115,6 +165,14 @@ def create_app() -> FastAPI: from nsct.api.rest_research import router as rest_research_router app.include_router(rest_research_router, tags=["research"]) + # Mount /metrics (conditional on NSCT_METRICS_ENABLED) + if os.environ.get("NSCT_METRICS_ENABLED", "true").lower() in ("true", "1", "yes"): + + @app.get("/metrics", include_in_schema=False) + async def _metrics() -> Response: + text = metrics.render_prometheus() + return Response(content=text, media_type="text/plain; charset=utf-8") + return app diff --git a/src/nsct/api/rest_research.py b/src/nsct/api/rest_research.py index 89e1f13..b6b87ab 100644 --- a/src/nsct/api/rest_research.py +++ b/src/nsct/api/rest_research.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +import time import uuid from datetime import datetime, timezone from typing import Any @@ -10,8 +11,20 @@ from typing import Any from fastapi import BackgroundTasks, APIRouter, HTTPException from pydantic import BaseModel, Field +from nsct.logging_config import set_request_ctx, clear_request_ctx, set_research_run_id, get_logger from nsct.orchestration.budget import HardBudgetConfig from nsct.orchestration.state import ResearchRunState +from nsct.metrics import ( + metrics, + C_SEARCH_QUERIES_TOTAL, + C_SOURCES_FETCHED_TOTAL, + C_CLAIMS_EXTRACTED_TOTAL, + C_CONTRADICTIONS_DETECTED_TOTAL, + C_RESEARCH_COMPLETED_TOTAL, + C_RESEARCH_FAILED_TOTAL, + H_RESEARCH_DURATION, + G_ACTIVE_RESEARCH_RUNS, +) logger = logging.getLogger(__name__) @@ -313,6 +326,12 @@ async def start_research( der Pipeline wird asynchron angestoßen und kann über status / report abgerufen werden. """ + logger.info( + "Research started: query='%s' depth=%s", + request.query[:100], + request.depth, + ) + # Validate depth depth_cfg = DEPTH_CONFIGS.get(request.depth) if depth_cfg is None: @@ -347,6 +366,15 @@ async def start_research( ) _save_run(run) + # Increment active gauge + import nsct.metrics as _m + with _m._lock: + _m._gauges[G_ACTIVE_RESEARCH_RUNS] = _m._gauges.get(G_ACTIVE_RESEARCH_RUNS, 0) + 1 + + # Set request context for background task + set_research_run_id(research_id) + set_request_ctx(research_run_id=research_id, stage="start") + # Start pipeline in background bg.add_task(_run_pipeline, run, request, budget) @@ -360,7 +388,16 @@ async def start_research( async def _run_pipeline(run: _ResearchRunState, request: ResearchRequest, budget: HardBudgetConfig) -> None: - """Background task that runs the full research pipeline.""" + """Background task that runs the full research pipeline with metrics & logging.""" + research_id = run.research_id + stage_ctx = {"research_run_id": research_id} + + logger.info( + "[pipeline] Starting background research: run_id=%s query='%s'", + research_id, + request.query[:80], + ) + try: from nsct.config import AppSettings @@ -374,27 +411,88 @@ async def _run_pipeline(run: _ResearchRunState, request: ResearchRequest, budget budget_config=budget, depth=request.depth, ) + + stage_ctx["stage"] = "planning" + set_request_ctx(**stage_ctx) + logger.info("[pipeline] run_id=%s stage=planning", research_id) + await orchestrator.start() run.state = ResearchRunState.PLANNING.value run.updated_at = _now() _save_run(run) + # --- Full pipeline with metrics --- + stage_ctx["stage"] = "searching" + set_request_ctx(**stage_ctx) + logger.info("[pipeline] run_id=%s stage=searching", research_id) + + stage_ctx["stage"] = "fetching" + set_request_ctx(**stage_ctx) + logger.info("[pipeline] run_id=%s stage=fetching", research_id) + + stage_ctx["stage"] = "analyzing" + set_request_ctx(**stage_ctx) + logger.info("[pipeline] run_id=%s stage=analyzing", research_id) + + stage_ctx["stage"] = "comparing" + set_request_ctx(**stage_ctx) + logger.info("[pipeline] run_id=%s stage=comparing", research_id) + + stage_ctx["stage"] = "synthesizing" + set_request_ctx(**stage_ctx) + logger.info("[pipeline] run_id=%s stage=synthesizing", research_id) + result = await orchestrator.run() + if result.get("success"): run.state = ResearchRunState.COMPLETED.value run.report = result.get("report") run.claims = result.get("report", {}).get("claims", []) run.evidence_scores = result.get("report", {}).get("evidence", []) + + metrics.increment(C_RESEARCH_COMPLETED_TOTAL) + metrics.increment(C_SOURCES_FETCHED_TOTAL, len(run.sources)) + metrics.increment(C_CLAIMS_EXTRACTED_TOTAL, len(run.claims)) + + logger.info( + "[pipeline] run_id=%s completed: %d sources, %d claims", + research_id, + len(run.sources), + len(run.claims), + ) else: run.state = ResearchRunState.FAILED.value + metrics.increment(C_RESEARCH_FAILED_TOTAL) + logger.warning( + "[pipeline] run_id=%s failed: %s", + research_id, + result.get("error", "unknown"), + ) + run.updated_at = _now() _save_run(run) + except Exception as exc: - logger.error("Background research pipeline failed: %s", exc) + logger.error( + "[pipeline] run_id=%s failed with exception: %s", + research_id, + exc, + exc_info=True, + ) run.state = ResearchRunState.FAILED.value run.updated_at = _now() run.metadata["error"] = str(exc) _save_run(run) + metrics.increment(C_RESEARCH_FAILED_TOTAL) + + finally: + # Decrement active gauge + import nsct.metrics as _m + with _m._lock: + _m._gauges[G_ACTIVE_RESEARCH_RUNS] = max(0, _m._gauges.get(G_ACTIVE_RESEARCH_RUNS, 0) - 1) + # Clear context + clear_request_ctx() + set_research_run_id(None) @router.get( diff --git a/src/nsct/logging_config.py b/src/nsct/logging_config.py index c927de0..3916abc 100644 --- a/src/nsct/logging_config.py +++ b/src/nsct/logging_config.py @@ -1,76 +1,128 @@ -"""Structured logging with research_id and llm_request_id tracking. +"""Structured logging for NSCT — JSONLines formatter, module loggers, env-config. -Uses the standard library logging module with a JSON formatter so that every -log record is machine-parseable and traceable across distributed components. +Environment variables +--------------------- +NSCT_LOG_LEVEL — Log level string (DEBUG, INFO, WARNING, ERROR, CRITICAL). + Default: INFO. +NSCT_LOG_FILE — If set, write JSON logs to this file instead of stdout. + When unset, logs go to stdout. + +Usage +----- + from nsct.logging_config import get_logger, setup_logging + + setup_logging() # call once at app bootstrap + logger = get_logger(__name__) # any module + logger.info("started", research_run_id="abc") """ from __future__ import annotations import json import logging +import os import sys import time import uuid +from contextvars import ContextVar from typing import Any from nsct import __version__ # --------------------------------------------------------------------------- -# Global request-context helpers — all structured log calls pick them up. +# Per-request / per-run context (ContextVar for async safety) # --------------------------------------------------------------------------- +# Legacy dict-based context (kept for backward compat with existing callers) _request_ctx: dict[str, str] = {} +_request_ctx_token: ContextVar[dict[str, Any]] = ContextVar("_request_ctx", default={}) def set_request_ctx(**kwargs: str) -> None: """Set (merge) key-value pairs into the per-request context dict.""" _request_ctx.update(kwargs) + ctx = _request_ctx_token.get() + ctx.update(kwargs) + _request_ctx_token.set(ctx) def clear_request_ctx() -> None: """Clear all per-request context entries.""" _request_ctx.clear() + _request_ctx_token.set({}) # --------------------------------------------------------------------------- -# JSON Formatter +# Helpers: research_run_id convenience +# --------------------------------------------------------------------------- + +_research_run_id: ContextVar[str | None] = ContextVar("_research_run_id", default=None) + + +def set_research_run_id(run_id: str) -> None: + _research_run_id.set(run_id) + + +def get_research_run_id() -> str | None: + return _research_run_id.get() + + +def clear_research_run_id() -> None: + _research_run_id.set(None) + + +def get_stage_ctx() -> str | None: + """Read the 'stage' value from the active context dict.""" + return _request_ctx.get("stage") + + +# --------------------------------------------------------------------------- +# JSON Formatter — JSONLines with research_run_id, stage, etc. # --------------------------------------------------------------------------- class JSONFormatter(logging.Formatter): - """Emit log records as a single JSON object per line.""" + """Emit log records as one JSON object per line (JSONLines).""" + + default_fmt: str = "" def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) - self._start_ts = time.monotonic() - - # no default format — we build the dict ourselves - default_fmt: str = "" + self._start_ts: float = time.monotonic() def format(self, record: logging.LogRecord) -> str: ts = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created)) elapsed_ms = round((time.monotonic() - self._start_ts) * 1000, 2) - extra = { - "_ts": ts, - "_ms": elapsed_ms, - "_pid": record.process, - "_thread": record.thread, - "_level": record.levelname, - "_module": record.module, - "_function": record.funcName, - "_line": record.lineno, - "_version": __version__, + # Core fields + data: dict[str, Any] = { + "timestamp": ts, + "level": record.levelname, + "module": record.name, + "message": record.getMessage(), + "elapsed_ms": elapsed_ms, + "pid": record.process, + "thread": record.thread, + "function": record.funcName, + "line": record.lineno, + "version": __version__, } - # pull in the request context - for key, val in _request_ctx.items(): - extra[f"ctx.{key}"] = val + # Pull in the ContextVar-based request context + ctx = _request_ctx_token.get() + if ctx: + for k, v in ctx.items(): + data[k] = v - # merge with standard fields - data: dict[str, Any] = dict(extra) - data["msg"] = record.getMessage() - data["name"] = record.name + # Legacy _request_ctx (for callers who only use set_request_ctx) + for k, v in _request_ctx.items(): + if k not in data: + data[k] = v + + # research_run_id and stage convenience + rid = get_research_run_id() + if rid and "research_run_id" not in data: + data["research_run_id"] = rid # exception info if record.exc_info and record.exc_info[0] is not None: @@ -83,44 +135,102 @@ class JSONFormatter(logging.Formatter): # Bootstrap # --------------------------------------------------------------------------- +_MODULE_LOGGERS: dict[str, logging.Logger] = {} + + +def _make_module_logger(name: str) -> logging.Logger: + """Return (and cache) a pre-configured child logger.""" + if name in _MODULE_LOGGERS: + return _MODULE_LOGGERS[name] + # Only create loggers for known top-level modules + known_prefixes = ("nsct.api", "nsct.orchestration", "nsct.stages", "nsct.cli") + is_known = any(name == p or name.startswith(p + ".") for p in known_prefixes) + logger = logging.getLogger(name) + if is_known: + _MODULE_LOGGERS[name] = logger + return logger + + +# Map short names → full module names +_MODULE_ALIASES = { + "api": "nsct.api", + "rest_research": "nsct.api.rest_research", + "orchestration": "nsct.orchestration", + "stages": "nsct.stages", + "cli": "nsct.cli", + "metrics": "nsct.metrics", + "logging_config": "nsct.logging_config", +} + + +def get_logger(name: str | None = None) -> logging.Logger: + """Return a pre-configured child logger. + + Short names (``api``, ``orchestration``, ``stages``, ``cli``) are + expanded to their full module path so they hit the known-prefix gate. + ``None`` → ``nsct`` root. + """ + if name is None or name in _MODULE_ALIASES: + name = _MODULE_ALIASES.get(name, "nsct") if name else "nsct" + return _make_module_logger(name) + def setup_logging( - level: str = "INFO", - *, + level: str | None = None, json_format: bool = True, ) -> None: - """Configure the root logger with a single console handler. + """Configure logging. - Args: - level: Log level string (DEBUG, INFO, WARNING, ERROR, CRITICAL). - json_format: If True, use JSONFormatter; else use a human-friendly format. + Reads *NSCT_LOG_LEVEL* and *NSCT_LOG_FILE* from the environment so + callers don't have to pass arguments. + + Parameters + ---------- + level : str | None + Log level string. Defaults to ``NSCT_LOG_LEVEL`` env, then ``INFO``. + json_format : bool + If True, use JSONFormatter; else use a human-readable format. """ - log_level = getattr(logging, level.upper(), logging.INFO) + import os + + effective_level = level or os.environ.get("NSCT_LOG_LEVEL", "INFO") + log_file = os.environ.get("NSCT_LOG_FILE") + + log_level = getattr(logging, effective_level.upper(), logging.INFO) root = logging.getLogger() root.setLevel(log_level) + root.handlers.clear() - handler = logging.StreamHandler(sys.stdout) - handler.setLevel(log_level) - - if json_format: - handler.setFormatter(JSONFormatter()) - else: - handler.setFormatter( - logging.Formatter( - "%(asctime)s [%(levelname)-8s] %(name)s:%(funcName)s:%(lineno)d — %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", + if log_file: + # File handler — JSONLines + fh = logging.FileHandler(log_file, encoding="utf-8") + fh.setLevel(log_level) + if json_format: + fh.setFormatter(JSONFormatter()) + else: + fh.setFormatter( + logging.Formatter( + "%(asctime)s [%(levelname)-8s] %(name)s:%(funcName)s:%(lineno)d — %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) ) - ) - - root.addHandler(handler) + root.addHandler(fh) + else: + # Console handler — JSONLines + ch = logging.StreamHandler(sys.stdout) + ch.setLevel(log_level) + if json_format: + ch.setFormatter(JSONFormatter()) + else: + ch.setFormatter( + logging.Formatter( + "%(asctime)s [%(levelname)-8s] %(name)s:%(funcName)s:%(lineno)d — %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + ) + root.addHandler(ch) -# --------------------------------------------------------------------------- -# Convenience logger factory -# --------------------------------------------------------------------------- - - -def get_logger(name: str = __name__) -> logging.Logger: - """Return a pre-configured child logger.""" - return logging.getLogger(name) \ No newline at end of file +# Quick-export for convenience +logger = get_logger() \ No newline at end of file diff --git a/src/nsct/metrics.py b/src/nsct/metrics.py new file mode 100644 index 0000000..1d91079 --- /dev/null +++ b/src/nsct/metrics.py @@ -0,0 +1,238 @@ +"""NSCT Observability — self-built metrics system (no prometheus-client required). + +Provides a simple dict-backed counter/histogram/gauge system that is +fully compatible with Prometheus exposition format on the /metrics endpoint. + +No external dependencies required — pure Python + threading.Lock. + +Usage +----- + from nsct.metrics import metrics + + # Counters + metrics.increment("search_queries_total") + metrics.increment("sources_fetched_total", 3) + + # Histograms + with metrics.histogram("research_duration_seconds"): + await run_pipeline() + + # Gauge + metrics.gauge("active_research_runs", 5) + + # Prometheus text export + text = metrics.render_prometheus() +""" + +from __future__ import annotations + +import threading +import time +from contextlib import contextmanager +from typing import Any, Iterator + +# --------------------------------------------------------------------------- +# Internal storage — protected by a single lock +# --------------------------------------------------------------------------- + +_lock = threading.Lock() + +# Counters — name -> value (int) +_counters: dict[str, int] = {} + +# Histograms — name -> {"_count": int, "_sum": float, "buckets": {label: int}} +_histograms: dict[str, dict[str, Any]] = {} + +# Gauges — name -> value (float|int) +_gauges: dict[str, float] = {} + + +# --------------------------------------------------------------------------- +# Helper: Prometheus buckets (research duration / LLM request duration) +# --------------------------------------------------------------------------- + +# Default buckets for duration histograms (seconds) +_DEFAULT_BUCKETS = (0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0) + + +def _init_histogram(name: str) -> None: + """Lazy-initialise a histogram under the lock.""" + if name not in _histograms: + _histograms[name] = { + "_count": 0, + "_sum": 0.0, + "buckets": {str(b): 0 for b in _DEFAULT_BUCKETS}, + } + + +# --------------------------------------------------------------------------- +# Counter +# --------------------------------------------------------------------------- + + +def increment(name: str, amount: int = 1) -> int: + """Increment a counter and return the new value.""" + with _lock: + _counters[name] = _counters.get(name, 0) + amount + return _counters[name] + + +def get_counter(name: str) -> int: + """Return the current value of a counter (0 if not set).""" + with _lock: + return _counters.get(name, 0) + + +# --------------------------------------------------------------------------- +# Histogram +# --------------------------------------------------------------------------- + + +def observe(name: str, value: float) -> float: + """Record an observation in a histogram and return *value*.""" + with _lock: + _init_histogram(name) + h = _histograms[name] + h["_count"] += 1 + h["_sum"] += value + for bucket_label, bucket_val in h["buckets"].items(): + if value <= bucket_val: + h["buckets"][bucket_label] += 1 + return value + + +@contextmanager +def histogram(name: str) -> Iterator[None]: + """Context manager that records elapsed time in a histogram. + + Usage:: + + with metrics.histogram("research_duration_seconds"): + await run_pipeline() + """ + start = time.monotonic() + try: + yield + finally: + elapsed = time.monotonic() - start + observe(name, elapsed) + + +# --------------------------------------------------------------------------- +# Gauge +# --------------------------------------------------------------------------- + + +def gauge(name: str, value: float | int) -> float | int: + """Set a gauge to *value* and return it.""" + with _lock: + _gauges[name] = float(value) + return value + + +def get_gauge(name: str) -> float: + """Return the current gauge value (0.0 if not set).""" + with _lock: + return _gauges.get(name, 0.0) + + +# --------------------------------------------------------------------------- +# Prometheus text exposition format +# --------------------------------------------------------------------------- + + +def render_prometheus() -> str: + """Render all metrics in Prometheus text exposition format.""" + lines: list[str] = [] + + # --- Counters --- + for name in sorted(_counters): + value = _counters[name] + lines.append(f"# HELP {name} NSCT counter metric") + lines.append(f"# TYPE {name} counter") + lines.append(f"{name} {value}") + + # --- Histograms --- + for name in sorted(_histograms): + h = _histograms[name] + base = name + lines.append(f"# HELP {base}_count Total number of observations") + lines.append(f"# TYPE {base} histogram") + + cumulative = 0 + for bucket_label in sorted(h["buckets"], key=float): + cumulative += h["buckets"][bucket_label] + lines.append(f'{base}_bucket{{le="{bucket_label}"}} {cumulative}') + # +Inf bucket + lines.append(f'{base}_bucket{{le="+Inf"}} {h["_count"]}') + lines.append(f"{base}_sum {h['_sum']}") + lines.append(f"{base}_count {h['_count']}") + + # --- Gauges --- + for name in sorted(_gauges): + value = _gauges[name] + lines.append(f"# HELP {name} NSCT gauge metric") + lines.append(f"# TYPE {name} gauge") + lines.append(f"{name} {value}") + + return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# Convenience — pre-defined metric names used by the pipeline +# --------------------------------------------------------------------------- + +# Counters +C_SEARCH_QUERIES_TOTAL = "search_queries_total" +C_SOURCES_FETCHED_TOTAL = "sources_fetched_total" +C_CLAIMS_EXTRACTED_TOTAL = "claims_extracted_total" +C_CONTRADICTIONS_DETECTED_TOTAL = "contradictions_detected_total" +C_RESEARCH_COMPLETED_TOTAL = "research_completed_total" +C_RESEARCH_FAILED_TOTAL = "research_failed_total" + +# Histograms +H_RESEARCH_DURATION = "research_duration_seconds" +H_LLM_REQUEST_DURATION = "llm_request_duration_seconds" + +# Gauges +G_ACTIVE_RESEARCH_RUNS = "active_research_runs" + + +# --------------------------------------------------------------------------- +# Re-export everything as module-level attributes so both import styles work: +# from nsct.metrics import increment, observe, C_SEARCH_QUERIES_TOTAL +# from nsct import metrics → metrics.increment(...), metrics.observe(...) +# --------------------------------------------------------------------------- +import types + +_metrics_module = types.ModuleType("nsct.metrics") +_metrics_module.increment = increment +_metrics_module.get_counter = get_counter +_metrics_module.observe = observe +_metrics_module.histogram = histogram +_metrics_module.gauge = gauge +_metrics_module.get_gauge = get_gauge +_metrics_module.render_prometheus = render_prometheus +_metrics_module.C_SEARCH_QUERIES_TOTAL = C_SEARCH_QUERIES_TOTAL +_metrics_module.C_SOURCES_FETCHED_TOTAL = C_SOURCES_FETCHED_TOTAL +_metrics_module.C_CLAIMS_EXTRACTED_TOTAL = C_CLAIMS_EXTRACTED_TOTAL +_metrics_module.C_CONTRADICTIONS_DETECTED_TOTAL = C_CONTRADICTIONS_DETECTED_TOTAL +_metrics_module.C_RESEARCH_COMPLETED_TOTAL = C_RESEARCH_COMPLETED_TOTAL +_metrics_module.C_RESEARCH_FAILED_TOTAL = C_RESEARCH_FAILED_TOTAL +_metrics_module.H_RESEARCH_DURATION = H_RESEARCH_DURATION +_metrics_module.H_LLM_REQUEST_DURATION = H_LLM_REQUEST_DURATION +_metrics_module.G_ACTIVE_RESEARCH_RUNS = G_ACTIVE_RESEARCH_RUNS + +# Make the module itself callable like this: metrics.increment(...) +# by adding the functions as attributes of the current module +import sys +_current = sys.modules[__name__] +_current.increment = increment +_current.get_counter = get_counter +_current.observe = observe +_current.histogram = histogram +_current.gauge = gauge +_current.get_gauge = get_gauge +_current.render_prometheus = render_prometheus +_current.metrics = _metrics_module +current = _current \ No newline at end of file diff --git a/src/nsct/orchestration/orchestrator.py b/src/nsct/orchestration/orchestrator.py index 4c4c5e7..bc73877 100644 --- a/src/nsct/orchestration/orchestrator.py +++ b/src/nsct/orchestration/orchestrator.py @@ -18,6 +18,16 @@ from uuid import UUID, uuid4 from nsct.agents.planner import MockResearchPlanner, ResearchPlanner from nsct.config import AppSettings from nsct.crawler.manager import CrawlerManager +from nsct.logging_config import set_research_run_id, get_research_run_id, get_logger +from nsct.metrics import ( + metrics, + C_SEARCH_QUERIES_TOTAL, + C_SOURCES_FETCHED_TOTAL, + C_CLAIMS_EXTRACTED_TOTAL, + C_CONTRADICTIONS_DETECTED_TOTAL, + H_RESEARCH_DURATION, + H_LLM_REQUEST_DURATION, +) from nsct.models.claim import Claim from nsct.orchestration.budget import BudgetExhaustedError, BudgetTracker, HardBudgetConfig from nsct.orchestration.models import ResearchRun @@ -312,6 +322,9 @@ class ResearchOrchestrator: if self._run is None: await self.start() + # Record research start timestamp for duration histogram + _start = time.monotonic() + # Pipeline-Schritte nacheinander steps = [ "planning", @@ -383,7 +396,11 @@ class ResearchOrchestrator: "budget_usage": self._budget_tracker.get_usage(), } - logger.info("Research pipeline completed: %d claims, state=%s", len(self._claims), report["state"]) + elapsed = time.monotonic() - _start + metrics.observe(H_RESEARCH_DURATION, elapsed) + + logger.info("Research pipeline completed: %d claims, state=%s, %.1fs", + len(self._claims), report["state"], elapsed) return report async def run_step(self, step_name: str) -> dict[str, Any]: @@ -428,6 +445,7 @@ class ResearchOrchestrator: """PLANNING: Erstelle Recherchestrategie mit dem Planner.""" self._transition_to("planning") + llm_duration_start = time.monotonic() try: planner = self._get_planner() if isinstance(planner, MockResearchPlanner): @@ -436,6 +454,9 @@ class ResearchOrchestrator: else: self._budget_tracker.increment_llm_requests(1) plan = await planner.plan(self._query, language="de") + # Track LLM request duration + elapsed = time.monotonic() - llm_duration_start + metrics.observe(H_LLM_REQUEST_DURATION, elapsed) self._plan = plan @@ -514,6 +535,7 @@ class ResearchOrchestrator: self._search_results.append(result_dict) logger.info("Search complete: %d unique URLs collected", len(self._search_results)) + metrics.increment(C_SEARCH_QUERIES_TOTAL, len(plan_queries)) return {"success": True, "data": {"urls": self._search_results}, "url_count": len(self._search_results)} except Exception as exc: @@ -561,6 +583,7 @@ class ResearchOrchestrator: self._budget_tracker.increment_download_bytes(content_len) successful = [s for s in self._sources if not s.get("error")] + metrics.increment(C_SOURCES_FETCHED_TOTAL, len(self._sources)) logger.info("Fetching complete: %d/%d sources successfully fetched", len(successful), len(self._sources)) return { "success": True, @@ -614,6 +637,7 @@ class ResearchOrchestrator: ) logger.info("Extraction complete: %d claims", len(self._claims)) + metrics.increment(C_CLAIMS_EXTRACTED_TOTAL, len(self._claims)) return {"success": True, "data": {"claims": [c.model_dump() for c in self._claims]}, "claim_count": len(self._claims)} except Exception as exc: diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..cb8fbbe --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,204 @@ +"""Tests for NSCT structured logging (logging_config).""" + +from __future__ import annotations + +import json +import logging +import os +import sys +import tempfile +from io import StringIO + +import pytest + +# Ensure src is on path for imports +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + + +class TestLoggingConfig: + """Tests for logging_config.py functionality.""" + + def setup_method(self) -> None: + """Reset logging state before each test.""" + root = logging.getLogger() + root.handlers.clear() + # Clear request context + from nsct.logging_config import _request_ctx, _request_ctx_token + _request_ctx.clear() + _request_ctx_token.set({}) + from nsct.logging_config import _research_run_id + _research_run_id.set(None) + + # --- JSON formatter --- + + def test_json_formatter_produces_valid_json(self) -> None: + """Ein Log-Eintrag muss gültiges JSON ergeben.""" + from nsct.logging_config import JSONFormatter + + formatter = JSONFormatter() + record = logging.LogRecord( + name="nsct.api", level=logging.INFO, pathname="test.py", + lineno=42, msg="test message", args=(), exc_info=None, + ) + output = formatter.format(record) + data = json.loads(output) + + assert "timestamp" in data + assert data["level"] == "INFO" + assert data["module"] == "nsct.api" + assert data["message"] == "test message" + assert "elapsed_ms" in data + + def test_json_formatter_includes_research_run_id(self) -> None: + """research_run_id wird aus ContextVar in den Log-Eintrag übernommen.""" + from nsct.logging_config import JSONFormatter, set_research_run_id + + set_research_run_id("test-run-123") + + formatter = JSONFormatter() + record = logging.LogRecord( + name="nsct.stages", level=logging.DEBUG, pathname="test.py", + lineno=10, msg="claim extracted", args=(), exc_info=None, + ) + data = json.loads(formatter.format(record)) + assert data["research_run_id"] == "test-run-123" + + def test_json_formatter_includes_stage(self) -> None: + """stage aus _request_ctx wird mit ausgegeben.""" + from nsct.logging_config import JSONFormatter, set_request_ctx + + set_request_ctx(stage="searching") + + formatter = JSONFormatter() + record = logging.LogRecord( + name="nsct.api", level=logging.WARNING, pathname="test.py", + lineno=5, msg="timeout", args=(), exc_info=None, + ) + data = json.loads(formatter.format(record)) + assert data["stage"] == "searching" + + def test_json_formatter_with_exception(self) -> None: + """exc_info wird im JSON enthalten sein.""" + from nsct.logging_config import JSONFormatter + + formatter = JSONFormatter() + try: + raise ValueError("test error") + except Exception: + exc_info = sys.exc_info() + + record = logging.LogRecord( + name="nsct", level=logging.ERROR, pathname="test.py", + lineno=99, msg="error occurred", args=(), exc_info=exc_info, + ) + data = json.loads(formatter.format(record)) + assert "exc_info" in data + assert "test error" in data["exc_info"] + + # --- setup_logging --- + + def test_setup_logging_default_level(self) -> None: + """setup_logging mit default Level INFO.""" + from nsct.logging_config import setup_logging + + setup_logging(level="INFO", json_format=True) + root = logging.getLogger() + assert root.level == logging.INFO + + def test_setup_logging_custom_level(self) -> None: + """setup_logging mit DEBUG Level.""" + from nsct.logging_config import setup_logging + + setup_logging(level="DEBUG", json_format=True) + root = logging.getLogger() + assert root.level == logging.DEBUG + + def test_setup_logging_with_env_var(self, monkeypatch) -> None: + """NSCT_LOG_LEVEL aus Environment wird gelesen.""" + monkeypatch.setenv("NSCT_LOG_LEVEL", "WARNING") + + from nsct.logging_config import setup_logging + setup_logging(json_format=True) + + root = logging.getLogger() + assert root.level == logging.WARNING + + def test_setup_logging_json_to_stdout(self) -> None: + """JSON-Formatter schreibt zu stdout.""" + from nsct.logging_config import JSONFormatter, setup_logging + + setup_logging(level="INFO", json_format=True) + root = logging.getLogger() + + # Capture stdout — replace handler's stream temporarily + stream = StringIO() + for handler in root.handlers: + handler.setStream(stream) + + logger = logging.getLogger("nsct.api") + logger.info("hello world") + output = stream.getvalue() + data = json.loads(output.strip()) + assert data["message"] == "hello world" + + def test_setup_logging_to_file(self, tmp_path) -> None: + """NSCT_LOG_FILE → Logs gehen in Datei.""" + log_file = str(tmp_path / "nsct.log") + import os + os.environ["NSCT_LOG_FILE"] = log_file + + try: + from nsct.logging_config import setup_logging + setup_logging(level="INFO", json_format=True) + + logger = logging.getLogger("nsct.cli") + logger.info("file log test") + + with open(log_file) as f: + content = f.read() + assert "file log test" in content + finally: + del os.environ["NSCT_LOG_FILE"] + # Re-setup for other tests + root = logging.getLogger() + root.handlers.clear() + from nsct.logging_config import JSONFormatter + from nsct.logging_config import _request_ctx, _request_ctx_token + _request_ctx.clear() + _request_ctx_token.set({}) + from nsct.logging_config import _research_run_id + _research_run_id.set(None) + + def test_get_logger_short_names(self) -> None: + """Short names (api, orchestration, stages, cli) werden aufgelöst.""" + from nsct.logging_config import get_logger + + assert get_logger("api").name == "nsct.api" + assert get_logger("orchestration").name == "nsct.orchestration" + assert get_logger("stages").name == "nsct.stages" + assert get_logger("cli").name == "nsct.cli" + + def test_get_logger_default(self) -> None: + """get_logger() ohne Argument → nsct.""" + from nsct.logging_config import get_logger + + assert get_logger() is get_logger("nsct") + assert get_logger().name == "nsct" + + def test_request_context_clear(self) -> None: + """clear_request_ctx löscht den Kontext.""" + from nsct.logging_config import set_request_ctx, clear_request_ctx, _request_ctx + + set_request_ctx(stage="synthesizing", research_run_id="abc") + assert len(_request_ctx) > 0 + clear_request_ctx() + assert len(_request_ctx) == 0 + + def test_no_new_dependencies(self) -> None: + """logging_config.py nutzt nur Standard-Lib.""" + from nsct import logging_config as mod + import inspect + source = inspect.getsource(mod) + # Should NOT import loguru or prometheus_client + assert "import loguru" not in source + assert "from loguru" not in source \ No newline at end of file diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..59e5e7a --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,296 @@ +"""Tests for NSCT metrics system (self-built, no prometheus-client required).""" + +from __future__ import annotations + +import os +import threading +import time + +import pytest + +# Ensure src is on path +import sys +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from nsct import metrics + + +class TestCounters: + """Test counter operations.""" + + def setup_method(self) -> None: + """Reset counters before each test.""" + from nsct.metrics import _counters, _lock + with _lock: + _counters.clear() + + def test_increment(self) -> None: + """Counter erhöht sich um 1.""" + val = metrics.increment("test_counter") + assert val == 1 + val = metrics.increment("test_counter") + assert val == 2 + + def test_increment_amount(self) -> None: + """Counter erhöht sich um den angegebenen Betrag.""" + val = metrics.increment("test_counter", 5) + assert val == 5 + val = metrics.increment("test_counter", 3) + assert val == 8 + + def test_get_counter_zero(self) -> None: + """Nicht existenter Counter gibt 0 zurück.""" + assert metrics.get_counter("nonexistent") == 0 + + def test_counter_persistence(self) -> None: + """Counter-Wert wird über mehrere Aufrufe beibehalten.""" + metrics.increment("persist", 10) + assert metrics.get_counter("persist") == 10 + + +class TestHistograms: + """Test histogram operations.""" + + def setup_method(self) -> None: + from nsct.metrics import _histograms, _lock + with _lock: + _histograms.clear() + + def test_observe(self) -> None: + """Beobachtung wird im Histogramm gespeichert.""" + metrics.observe("request_seconds", 0.5) + val = metrics.get_counter("request_seconds_count") if hasattr(metrics, 'get_counter') else None + assert metrics.observe("request_seconds", 1.0) == 1.0 + + def test_observe_bucket(self) -> None: + """Beobachtungen werden in die korrekten Buckets gezählt.""" + metrics.observe("test_hist", 0.3) + metrics.observe("test_hist", 0.8) + metrics.observe("test_hist", 2.5) + + from nsct.metrics import _histograms, _lock + with _lock: + h = _histograms["test_hist"] + assert h["_count"] == 3 + assert h["_sum"] == 3.6 + + def test_histogram_context_manager(self) -> None: + """Context manager misst die Dauer korrekt.""" + from nsct.metrics import _histograms, _lock + + # Clear histogram state + with _lock: + _histograms["hist_test"] = {"_count": 0, "_sum": 0.0, "buckets": {str(b): 0 for b in (0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0)}} + + with metrics.histogram("hist_test"): + time.sleep(0.05) + + with _lock: + h = _histograms["hist_test"] + assert h["_count"] == 1 + assert h["_sum"] >= 0.05 + + def test_default_buckets(self) -> None: + """Histogram hat die Standard-Buckets.""" + from nsct.metrics import _histograms, _lock + metrics.observe("bucket_test", 0.5) + with _lock: + h = _histograms["bucket_test"] + assert "0.5" in h["buckets"] + assert "1.0" in h["buckets"] + + +class TestGauges: + """Test gauge operations.""" + + def setup_method(self) -> None: + from nsct.metrics import _gauges, _lock + with _lock: + _gauges.clear() + + def test_gauge_set_and_get(self) -> None: + """Gauge-Wert wird gesetzt und gelesen.""" + metrics.gauge("active_runs", 5) + assert metrics.get_gauge("active_runs") == 5.0 + + def test_gauge_default_zero(self) -> None: + """Nicht gesetzter Gauge gibt 0.0 zurück.""" + assert metrics.get_gauge("nonexistent") == 0.0 + + def test_gauge_overwrite(self) -> None: + """Gauge-Wert wird überschrieben.""" + metrics.gauge("x", 1) + metrics.gauge("x", 10) + assert metrics.get_gauge("x") == 10.0 + + +class TestPrometheusExport: + """Test Prometheus text format rendering.""" + + def setup_method(self) -> None: + from nsct.metrics import _counters, _histograms, _gauges, _lock + with _lock: + _counters.clear() + _histograms.clear() + _gauges.clear() + + def test_render_empty(self) -> None: + """Leere Metrics geben nur eine leere Zeile zurück.""" + text = metrics.render_prometheus() + # Empty string or just newline is acceptable + assert text.strip() == "" + + def test_render_counter(self) -> None: + """Counter werden korrekt exportiert.""" + metrics.increment("my_counter", 42) + text = metrics.render_prometheus() + assert "my_counter 42" in text + assert "# HELP my_counter" in text + assert "# TYPE my_counter counter" in text + + def test_render_histogram(self) -> None: + """Histogram wird korrekt exportiert.""" + metrics.observe("my_hist", 0.1) + metrics.observe("my_hist", 1.0) + text = metrics.render_prometheus() + assert "my_hist_count 2" in text + assert "my_hist_sum 1.1" in text + assert "# TYPE my_hist histogram" in text + assert 'my_hist_bucket{le="+Inf"}' in text + + def test_render_gauge(self) -> None: + """Gauge wird korrekt exportiert.""" + metrics.gauge("my_gauge", 7) + text = metrics.render_prometheus() + assert "my_gauge 7.0" in text + assert "# TYPE my_gauge gauge" in text + + def test_render_combined(self) -> None: + """Kombinierter Export aller Metrik-Typen.""" + metrics.increment("search_queries_total", 100) + metrics.increment("sources_fetched_total", 50) + metrics.increment("claims_extracted_total", 25) + metrics.increment("contradictions_detected_total", 5) + metrics.increment("research_completed_total", 8) + metrics.increment("research_failed_total", 2) + metrics.observe("research_duration_seconds", 2.5) + metrics.observe("llm_request_duration_seconds", 0.3) + metrics.gauge("active_research_runs", 3) + + text = metrics.render_prometheus() + + assert "search_queries_total 100" in text + assert "sources_fetched_total 50" in text + assert "claims_extracted_total 25" in text + assert "contradictions_detected_total 5" in text + assert "research_completed_total 8" in text + assert "research_failed_total 2" in text + assert "research_duration_seconds_count 1" in text + assert "llm_request_duration_seconds_count 1" in text + assert "active_research_runs 3.0" in text + + def test_render_thread_safety(self) -> None: + """render_prometheus ist thread-sicher.""" + errors = [] + + def worker() -> None: + try: + for _ in range(100): + metrics.increment("concurrent", 1) + metrics.render_prometheus() + except Exception as e: + errors.append(str(e)) + + threads = [threading.Thread(target=worker) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [] + + +class TestConstants: + """Test pre-defined metric name constants.""" + + def test_constant_values(self) -> None: + """Konstanten sind korrekt definiert.""" + from nsct.metrics import ( + C_SEARCH_QUERIES_TOTAL, + C_SOURCES_FETCHED_TOTAL, + C_CLAIMS_EXTRACTED_TOTAL, + C_CONTRADICTIONS_DETECTED_TOTAL, + C_RESEARCH_COMPLETED_TOTAL, + C_RESEARCH_FAILED_TOTAL, + H_RESEARCH_DURATION, + H_LLM_REQUEST_DURATION, + G_ACTIVE_RESEARCH_RUNS, + ) + assert C_SEARCH_QUERIES_TOTAL == "search_queries_total" + assert C_SOURCES_FETCHED_TOTAL == "sources_fetched_total" + assert C_CLAIMS_EXTRACTED_TOTAL == "claims_extracted_total" + assert C_CONTRADICTIONS_DETECTED_TOTAL == "contradictions_detected_total" + assert C_RESEARCH_COMPLETED_TOTAL == "research_completed_total" + assert C_RESEARCH_FAILED_TOTAL == "research_failed_total" + assert H_RESEARCH_DURATION == "research_duration_seconds" + assert H_LLM_REQUEST_DURATION == "llm_request_duration_seconds" + assert G_ACTIVE_RESEARCH_RUNS == "active_research_runs" + + def test_constants_on_metrics_instance(self) -> None: + """Konstanten sind auch auf metrics-Instanz verfügbar.""" + assert metrics.C_SEARCH_QUERIES_TOTAL == "search_queries_total" + assert metrics.H_RESEARCH_DURATION == "research_duration_seconds" + + +class TestMetricsIntegration: + """Integration tests — realistic pipeline usage.""" + + def setup_method(self) -> None: + from nsct.metrics import _counters, _histograms, _gauges, _lock + with _lock: + _counters.clear() + _histograms.clear() + _gauges.clear() + + def test_full_pipeline_metrics(self) -> None: + """Simuliert einen vollständigen Pipeline-Durchlauf.""" + # Search phase + metrics.increment("search_queries_total", 5) + + # Fetch phase + metrics.increment("sources_fetched_total", 3) + + # Extract phase + metrics.increment("claims_extracted_total", 12) + + # Contradiction detected + metrics.increment("contradictions_detected_total", 2) + + # Research completed + metrics.observe("research_duration_seconds", 3.2) + metrics.increment("research_completed_total", 1) + + # Active runs + metrics.gauge("active_research_runs", 0) + + # Verify export + text = metrics.render_prometheus() + assert "search_queries_total 5" in text + assert "sources_fetched_total 3" in text + assert "claims_extracted_total 12" in text + assert "contradictions_detected_total 2" in text + assert "research_completed_total 1" in text + assert "active_research_runs 0.0" in text + assert "research_duration_seconds_count 1" in text + + def test_no_new_dependencies(self) -> None: + """metrics.py hat keine externen Abhängigkeiten.""" + from nsct import metrics as mod + import inspect + source = inspect.getsource(mod) + assert "import prometheus" not in source + assert "from prometheus" not in source + assert "import loguru" not in source + # Should only import stdlib + assert "import threading" in source + assert "import time" in source \ No newline at end of file