Add Vision image fetch failure observability

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))
 
 

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)

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]}")
 
 

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()