From 4ed3b619be4fac27862656e3e85a1fea9b907833 Mon Sep 17 00:00:00 2001 From: Gregor Klevze Date: Sat, 29 Aug 2026 07:56:35 +0200 Subject: [PATCH] Add Vision image fetch failure observability --- clip/main.py | 22 ++++++++- common/image_io.py | 76 ++++++++++++++++++++++++------ qdrant/main.py | 30 ++++++++++++ tests/test_image_io.py | 104 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 217 insertions(+), 15 deletions(-) create mode 100644 tests/test_image_io.py diff --git a/clip/main.py b/clip/main.py index 20480c8..c2e2fbd 100644 --- a/clip/main.py +++ b/clip/main.py @@ -1,6 +1,8 @@ from __future__ import annotations import os +import logging +import time from typing import List, Optional import torch @@ -9,7 +11,7 @@ from fastapi import FastAPI, HTTPException, UploadFile, File, Form from pydantic import BaseModel, Field import numpy as np -from common.image_io import fetch_url_bytes, bytes_to_pil, ImageLoadError +from common.image_io import fetch_url_bytes, bytes_to_pil, ImageLoadError, image_url_log_context MODEL_NAME = os.getenv("MODEL_NAME", "ViT-B-32") MODEL_PRETRAINED = os.getenv("MODEL_PRETRAINED", "openai") @@ -25,6 +27,7 @@ TAGS: List[str] = [ ] app = FastAPI(title="Skinbase CLIP Service", version="1.0.0") +logger = logging.getLogger("clip") model, _, preprocess = open_clip.create_model_and_transforms(MODEL_NAME, pretrained=MODEL_PRETRAINED) tokenizer = open_clip.get_tokenizer(MODEL_NAME) @@ -44,6 +47,21 @@ class EmbedRequest(BaseModel): pretrained: Optional[str] = None +def _log_image_load_failure(error: ImageLoadError, url: str, elapsed_ms: float) -> None: + fields = { + "event": "clip_image_fetch_failed", + "stage": error.stage, + "error_type": error.category, + "exception_class": error.original_exception_class or error.__class__.__name__, + "elapsed_ms": round(elapsed_ms, 1), + **image_url_log_context(url), + } + for key in ("remote_status", "content_type"): + if key in error.metadata: + fields[key] = error.metadata[key] + logger.warning("%s", " ".join(f"{key}={value!r}" for key, value in fields.items())) + + @app.get("/health") def health(): return {"status": "ok", "device": DEVICE, "model": MODEL_NAME, "pretrained": MODEL_PRETRAINED} @@ -143,10 +161,12 @@ async def analyze_file( def embed(req: EmbedRequest): if not req.url: raise HTTPException(400, "url is required") + started_at = time.perf_counter() try: data = fetch_url_bytes(req.url) return _embed_image_bytes(data, backend=req.backend, model_name=req.model, pretrained=req.pretrained) except ImageLoadError as e: + _log_image_load_failure(e, req.url, (time.perf_counter() - started_at) * 1000) raise HTTPException(400, str(e)) diff --git a/common/image_io.py b/common/image_io.py index 375ba8d..48a249f 100644 --- a/common/image_io.py +++ b/common/image_io.py @@ -1,6 +1,7 @@ from __future__ import annotations import io +import hashlib import ipaddress import socket from urllib.parse import urljoin, urlparse @@ -12,24 +13,57 @@ DEFAULT_MAX_BYTES = 50 * 1024 * 1024 # 50MB DEFAULT_MAX_REDIRECTS = 3 class ImageLoadError(Exception): - pass + """Image failure with stable internal diagnostics and unchanged message text.""" + + def __init__(self, message: str, *, category: str = "unknown", stage: str = "fetch", original_exception_class: str | None = None, metadata: dict[str, object] | None = None): + super().__init__(message) + self.category = category + self.stage = stage + self.original_exception_class = original_exception_class + self.metadata = metadata or {} + + +def image_url_log_context(url: str) -> dict[str, str]: + """Safe URL fields for logs; query strings, credentials and full paths stay private.""" + try: + parsed = urlparse(url) + result = {"host": parsed.hostname or "unknown", "path_hash": hashlib.sha256(parsed.path.encode("utf-8", "replace")).hexdigest()[:12]} + if parsed.scheme: + result["scheme"] = parsed.scheme.lower() + if parsed.port: + result["port"] = str(parsed.port) + return result + except Exception: + return {} + + +def _error(message: str, *, category: str, stage: str, cause: BaseException | None = None, metadata: dict[str, object] | None = None) -> ImageLoadError: + return ImageLoadError(message, category=category, stage=stage, original_exception_class=type(cause).__name__ if cause else None, metadata=metadata) def _validate_public_url(url: str) -> str: - parsed = urlparse(url) + try: + parsed = urlparse(url) + except ValueError as e: + raise _error(str(e), category="invalid_url", stage="validate", cause=e) from e if parsed.scheme not in ("http", "https"): - raise ImageLoadError("Only http and https URLs are allowed") + raise _error("Only http and https URLs are allowed", category="invalid_scheme", stage="validate") if not parsed.hostname: - raise ImageLoadError("URL must include a hostname") + raise _error("URL must include a hostname", category="missing_hostname", stage="validate") hostname = parsed.hostname.strip().lower() if hostname in {"localhost", "127.0.0.1", "::1"}: - raise ImageLoadError("Localhost URLs are not allowed") + raise _error("Localhost URLs are not allowed", category="blocked_address", stage="validate") try: - resolved = socket.getaddrinfo(hostname, parsed.port or (443 if parsed.scheme == "https" else 80), type=socket.SOCK_STREAM) + port = parsed.port + resolved = socket.getaddrinfo(hostname, port or (443 if parsed.scheme == "https" else 80), type=socket.SOCK_STREAM) + except ValueError as e: + raise _error(str(e), category="invalid_url", stage="validate", cause=e) from e except socket.gaierror as e: - raise ImageLoadError(f"Cannot resolve host: {e}") from e + raise _error(f"Cannot resolve host: {e}", category="dns_failure", stage="validate", cause=e) from e + except (TimeoutError, socket.timeout) as e: + raise _error("Host resolution timed out", category="connect_timeout", stage="validate", cause=e) from e for entry in resolved: address = entry[4][0] @@ -42,7 +76,7 @@ def _validate_public_url(url: str) -> str: or ip.is_reserved or ip.is_unspecified ): - raise ImageLoadError("URLs resolving to private or reserved addresses are not allowed") + raise _error("URLs resolving to private or reserved addresses are not allowed", category="blocked_address", stage="validate") return url @@ -56,11 +90,15 @@ def fetch_url_bytes(url: str, timeout: float = 10.0, max_bytes: int = DEFAULT_MA if 300 <= r.status_code < 400: location = r.headers.get("location") if not location: - raise ImageLoadError("Redirect response missing location header") + raise _error("Redirect response missing location header", category="redirect_missing_location", stage="fetch", metadata={"remote_status": r.status_code}) current_url = _validate_public_url(urljoin(current_url, location)) continue - r.raise_for_status() + try: + r.raise_for_status() + except requests.HTTPError as e: + category = "remote_http_4xx" if 400 <= r.status_code < 500 else "remote_http_5xx" if r.status_code >= 500 else "fetch_failure" + raise _error(str(e), category=category, stage="fetch", cause=e, metadata={"remote_status": r.status_code}) from e content_type = (r.headers.get("content-type") or "").lower() if content_type and not content_type.startswith("image/"): @@ -73,19 +111,29 @@ def fetch_url_bytes(url: str, timeout: float = 10.0, max_bytes: int = DEFAULT_MA continue total += len(chunk) if total > max_bytes: - raise ImageLoadError(f"Image exceeds max_bytes={max_bytes}") + raise _error(f"Image exceeds max_bytes={max_bytes}", category="image_too_large", stage="fetch", metadata={"remote_status": r.status_code}) buf.write(chunk) return buf.getvalue() - raise ImageLoadError(f"Too many redirects (>{DEFAULT_MAX_REDIRECTS})") + raise _error(f"Too many redirects (>{DEFAULT_MAX_REDIRECTS})", category="too_many_redirects", stage="fetch") except ImageLoadError: raise + except requests.exceptions.ConnectTimeout as e: + raise _error(f"Cannot fetch image url: {e}", category="connect_timeout", stage="fetch", cause=e) from e + except requests.exceptions.ReadTimeout as e: + raise _error(f"Cannot fetch image url: {e}", category="read_timeout", stage="fetch", cause=e) from e + except requests.exceptions.Timeout as e: + raise _error(f"Cannot fetch image url: {e}", category="read_timeout", stage="fetch", cause=e) from e + except requests.exceptions.TooManyRedirects as e: + raise _error(f"Cannot fetch image url: {e}", category="too_many_redirects", stage="fetch", cause=e) from e + except requests.exceptions.ConnectionError as e: + raise _error(f"Cannot fetch image url: {e}", category="connection_error", stage="fetch", cause=e) from e except Exception as e: - raise ImageLoadError(f"Cannot fetch image url: {e}") from e + raise _error(f"Cannot fetch image url: {e}", category="fetch_failure", stage="fetch", cause=e) from e def bytes_to_pil(data: bytes) -> Image.Image: try: img = Image.open(io.BytesIO(data)).convert("RGB") return img except Exception as e: - raise ImageLoadError(f"Cannot decode image: {e}") from e + raise _error(f"Cannot decode image: {e}", category="decode_failure", stage="decode", cause=e) diff --git a/qdrant/main.py b/qdrant/main.py index 3928eba..09bde00 100644 --- a/qdrant/main.py +++ b/qdrant/main.py @@ -189,34 +189,64 @@ def _col(name: Optional[str]) -> str: return name or COLLECTION_NAME +def _safe_clip_detail(response: httpx.Response) -> str: + """Extract only a bounded machine-readable category; never log response text.""" + try: + detail = response.json().get("detail") + except Exception: + return "non_json_response" + if not isinstance(detail, str): + return "structured_error" + import re + match = re.search(r"(?:error_type|category)[=: ]+([a-z_]+)", detail) + return match.group(1)[:40] if match else "upstream_error" + + +def _log_clip_failure(operation: str, *, elapsed_ms: float, status: int | None = None, error_type: str, safe_detail: str | None = None) -> None: + fields = {"event": "qdrant_clip_upstream_failed", "upstream": "clip", "operation": operation, "elapsed_ms": round(elapsed_ms, 1), "error_type": error_type} + if status is not None: + fields["upstream_status"] = status + if safe_detail is not None: + fields["safe_detail"] = safe_detail + logger.warning("%s", " ".join(f"{key}={value!r}" for key, value in fields.items())) + + async def _embed_url(url: str) -> List[float]: """Call the CLIP service to get an image embedding.""" + started_at = time.perf_counter() async with httpx.AsyncClient(timeout=30) as http: try: r = await http.post(f"{CLIP_URL}/embed", json={"url": url}) except httpx.RequestError as e: + _log_clip_failure("embed_url", elapsed_ms=(time.perf_counter() - started_at) * 1000, error_type=type(e).__name__) raise HTTPException(502, f"CLIP request failed: {str(e)}") if r.status_code >= 400: + _log_clip_failure("embed_url", elapsed_ms=(time.perf_counter() - started_at) * 1000, status=r.status_code, error_type="upstream_http_error", safe_detail=_safe_clip_detail(r)) raise HTTPException(502, f"CLIP /embed error: {r.status_code} {r.text[:200]}") try: return r.json()["vector"] except Exception: + _log_clip_failure("embed_url", elapsed_ms=(time.perf_counter() - started_at) * 1000, status=r.status_code, error_type="invalid_response", safe_detail=_safe_clip_detail(r)) raise HTTPException(502, f"CLIP /embed returned non-JSON: {r.status_code} {r.text[:200]}") async def _embed_bytes(data: bytes) -> List[float]: """Call the CLIP service to embed uploaded file bytes.""" + started_at = time.perf_counter() async with httpx.AsyncClient(timeout=30) as http: files = {"file": ("image", data, "application/octet-stream")} try: r = await http.post(f"{CLIP_URL}/embed/file", files=files) except httpx.RequestError as e: + _log_clip_failure("embed_file", elapsed_ms=(time.perf_counter() - started_at) * 1000, error_type=type(e).__name__) raise HTTPException(502, f"CLIP request failed: {str(e)}") if r.status_code >= 400: + _log_clip_failure("embed_file", elapsed_ms=(time.perf_counter() - started_at) * 1000, status=r.status_code, error_type="upstream_http_error", safe_detail=_safe_clip_detail(r)) raise HTTPException(502, f"CLIP /embed/file error: {r.status_code} {r.text[:200]}") try: return r.json()["vector"] except Exception: + _log_clip_failure("embed_file", elapsed_ms=(time.perf_counter() - started_at) * 1000, status=r.status_code, error_type="invalid_response", safe_detail=_safe_clip_detail(r)) raise HTTPException(502, f"CLIP /embed/file returned non-JSON: {r.status_code} {r.text[:200]}") diff --git a/tests/test_image_io.py b/tests/test_image_io.py new file mode 100644 index 0000000..1a4f5b4 --- /dev/null +++ b/tests/test_image_io.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import socket +import unittest +from unittest.mock import Mock, patch + +import requests + +from common.image_io import ImageLoadError, bytes_to_pil, fetch_url_bytes + + +class _Response: + def __init__(self, status_code=200, *, headers=None, body=b"image-bytes"): + self.status_code = status_code + self.headers = headers or {"content-type": "image/webp"} + self.body = body + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def iter_content(self, chunk_size=0): + yield self.body + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"HTTP {self.status_code}") + + +class ImageIoClassificationTests(unittest.TestCase): + def assert_category(self, operation, category): + with self.assertRaises(ImageLoadError) as raised: + operation() + self.assertEqual(raised.exception.category, category) + + @patch("common.image_io.socket.getaddrinfo") + def test_invalid_url_scheme(self, _getaddrinfo): + self.assert_category(lambda: fetch_url_bytes("ftp://cdn.example/a.webp"), "invalid_scheme") + + @patch("common.image_io.socket.getaddrinfo") + def test_missing_hostname(self, _getaddrinfo): + self.assert_category(lambda: fetch_url_bytes("https:///a.webp"), "missing_hostname") + + @patch("common.image_io.socket.getaddrinfo", side_effect=socket.gaierror("NXDOMAIN")) + def test_dns_failure(self, _getaddrinfo): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "dns_failure") + + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("127.0.0.1", 443))]) + def test_private_address_rejected(self, _getaddrinfo): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "blocked_address") + + @patch("common.image_io.requests.get", side_effect=requests.ConnectTimeout("connect")) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_connect_timeout(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "connect_timeout") + + @patch("common.image_io.requests.get", side_effect=requests.ReadTimeout("read")) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_read_timeout(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "read_timeout") + + @patch("common.image_io.requests.get", side_effect=requests.ConnectionError("connection")) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_generic_connection_failure(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "connection_error") + + @patch("common.image_io.requests.get", return_value=_Response(302)) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_redirect_without_location(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "redirect_missing_location") + + @patch("common.image_io.requests.get", side_effect=[_Response(302, headers={"location": "https://cdn.example/a.webp"})] * 4) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_too_many_redirects(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "too_many_redirects") + + @patch("common.image_io.requests.get", return_value=_Response(404)) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_remote_404(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "remote_http_4xx") + + @patch("common.image_io.requests.get", return_value=_Response(500)) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_remote_500(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "remote_http_5xx") + + @patch("common.image_io.requests.get", return_value=_Response(headers={"content-type": "text/html"})) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_invalid_content_type(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp"), "invalid_content_type") + + @patch("common.image_io.requests.get", return_value=_Response(body=b"12345")) + @patch("common.image_io.socket.getaddrinfo", return_value=[(None, None, None, None, ("93.184.216.34", 443))]) + def test_max_bytes_exceeded(self, _getaddrinfo, _get): + self.assert_category(lambda: fetch_url_bytes("https://cdn.example/a.webp", max_bytes=4), "image_too_large") + + def test_decode_failure(self): + self.assert_category(lambda: bytes_to_pil(b"not-an-image"), "decode_failure") + + +if __name__ == "__main__": + unittest.main()