feat(observability): structured logging, metrics wiring, graceful degradation

- Item 5: Convert remaining printf-style logger call to structured logging
  (key=value format) in x402_middleware.py
- Item 6: Wire CACHE_HITS counter into cache get() and per-tier SCRAPE_LATENCY
  histogram into UltimateScraper scrape loop
- Item 7: Add circuit-breaker checks in extractor.py (Ollama) and
  routers/vision.py (OpenRouter) to fail fast and record success/failure
  when external services are down; attach recovery results to resilience
  registry for per-service circuit state

560 tests passing, ruff clean
This commit is contained in:
Crypto Rug Munch 2026-07-03 12:13:03 +02:00
parent 2970c15dbb
commit 5a21133b1d
5 changed files with 20 additions and 1 deletions

View file

@ -12,6 +12,8 @@ import json
import time
from collections import OrderedDict
from observability import CACHE_HITS
class ResponseCache:
"""Two-tier cache: local LRU + optional Redis persistence.
@ -44,6 +46,8 @@ class ResponseCache:
if entry["expires"] > now:
self._hits += 1
self._cache.move_to_end(key)
if CACHE_HITS:
CACHE_HITS.labels(cache_type="local").inc()
return entry["data"]
else:
del self._cache[key]

View file

@ -14,6 +14,8 @@ import json
import re
from typing import Any
from resilience import registry
class SchemaExtractor:
"""Extract structured JSON data from scraped markdown content.
@ -49,6 +51,7 @@ class SchemaExtractor:
merged = {**pattern_result, **llm_result}
return {k: v for k, v in merged.items() if v is not None and v != ""}
except Exception: # noqa: BLE001
registry.record_service_result("ollama", False)
pass
return pattern_result
@ -106,6 +109,8 @@ class SchemaExtractor:
async def _llm_extract(self, content: str, schema: dict[str, Any]) -> dict[str, Any]:
"""LLM-guided extraction. Returns dict on success, {"_error": msg} on failure."""
if not registry.is_service_allowed("ollama"):
return {"_error": "ollama circuit breaker open"}
import httpx
schema_str = json.dumps(schema, indent=2)
@ -136,7 +141,10 @@ class SchemaExtractor:
if json_match:
obj = json.loads(json_match.group(0))
if isinstance(obj, dict):
registry.record_service_result("ollama", True)
return obj
registry.record_service_result("ollama", True)
return {"_raw": response[:500]}
except (json.JSONDecodeError, ValueError) as e:
registry.record_service_result("ollama", False)
return {"_error": str(e)}

View file

@ -19,6 +19,7 @@ from fastapi import APIRouter, Body
from client import get_client
from deps import automator
from errors import ExternalServiceError, InvalidRequestError, PryError, ScrapeError
from resilience import registry
from settings import settings
logger = logging.getLogger(__name__)
@ -171,6 +172,8 @@ async def _call_vision_api(
key = _get_or_key()
if not key:
return None, None, "OPENROUTER_API_KEY not set"
if not registry.is_service_allowed("openrouter"):
return None, None, "OpenRouter circuit breaker open"
# Accept either data URI or raw base64
data_uri = image_b64 if image_b64.startswith("data:") else f"data:image/png;base64,{image_b64}"
@ -203,6 +206,7 @@ async def _call_vision_api(
)
resp.raise_for_status()
body = resp.json()
registry.record_service_result("openrouter", True)
return body["choices"][0]["message"]["content"], model, None

View file

@ -19,6 +19,7 @@ from typing import Any, ClassVar
from urllib.parse import urlparse
from errors import InvalidRequestError
from observability import SCRAPE_LATENCY
from resilience import registry
from url_guard import validate_url
@ -188,6 +189,8 @@ class UltimateScraper:
registry.record_domain_tier_result(domain, tier_name, success, latency)
if breaker_name:
registry.record_service_result(breaker_name, success)
if SCRAPE_LATENCY:
SCRAPE_LATENCY.labels(method=tier_name).observe(latency)
if success:
method = "premium_proxy" if tier_name == "premium_proxy" else tier_name

View file

@ -132,7 +132,7 @@ class X402Middleware:
await self.app(scope, receive, wrapped_send)
return
except (ValueError, KeyError, TypeError) as e:
logger.warning("x402_payment_decode_failed err=%s", e)
logger.warning("x402_payment_decode_failed", extra={"error": str(e)})
operation = next((op for ep, op in PAID_ENDPOINTS.items() if path.startswith(ep)), None)
if not operation or operation not in X402_PRICING: