加入 Ollama GCP failover 診斷與 unhealthy skip

services/ollama_service.py

@@ -10,6 +10,7 @@ import requests
 import json
 import logging
 import fnmatch
+from datetime import datetime, timedelta
 from typing import Optional, Dict, Any, List, Tuple
 from dataclasses import dataclass
 
@@ -172,6 +173,85 @@ def _reset_embedding_gcp_circuit() -> None:
     _embedding_gcp_failure_circuit.update({'blocked_until': 0.0, 'notice_ts': 0.0, 'tried': ()})
 
 
+def _host_label_for_embedding_health(host: str) -> str:
+    """Map an Ollama host URL to the host_health_probes label used by scheduler."""
+    if not host:
+        return ''
+    if '34.143.170.20:11434' in host or '192.168.0.110:11435' in host:
+        return 'Primary (GCP)'
+    if '34.21.145.224:11434' in host or '192.168.0.110:11436' in host:
+        return 'Secondary (GCP)'
+    return ''
+
+
+def _recent_embedding_host_unhealthy(host: str) -> bool:
+    """Skip known-bad GCP embedding runtimes using recent host_health_probes rows.
+
+    This guard is used only by background embedding paths that already disallow
+    111 fallback. It is deliberately fail-open: DB/read errors must not block
+    Ollama calls.
+    """
+    if not _env_flag('OLLAMA_EMBED_HOST_HEALTH_SKIP_ENABLED', True):
+        return False
+
+    host_label = _host_label_for_embedding_health(host)
+    if not host_label:
+        return False
+
+    try:
+        window_minutes = int(os.getenv('OLLAMA_EMBED_HOST_HEALTH_SKIP_WINDOW_MINUTES', '20'))
+    except (TypeError, ValueError):
+        window_minutes = 20
+    window_minutes = max(1, window_minutes)
+
+    try:
+        from sqlalchemy import text as sa_text
+        from database.manager import get_session
+
+        session = get_session()
+        try:
+            row = session.execute(
+                sa_text("""
+                    SELECT healthy, error_msg, probed_at
+                    FROM host_health_probes
+                    WHERE host_label = :host_label
+                    ORDER BY probed_at DESC
+                    LIMIT 1
+                """),
+                {'host_label': host_label},
+            ).fetchone()
+        finally:
+            session.close()
+    except Exception:
+        logger.debug("[Embed] host health skip fail-open for host=%s", host, exc_info=True)
+        return False
+
+    if not row:
+        return False
+
+    healthy, error_msg, probed_at = row[0], row[1], row[2]
+    if probed_at:
+        try:
+            now = datetime.now(probed_at.tzinfo) if getattr(probed_at, 'tzinfo', None) else datetime.now()
+            if now - probed_at > timedelta(minutes=window_minutes):
+                return False
+        except Exception:
+            logger.debug("[Embed] could not evaluate host health probe age for host=%s", host, exc_info=True)
+            return False
+
+    if bool(healthy):
+        return False
+
+    logger.warning(
+        "[Embed] skip recent unhealthy GCP embedding host=%s label=%s window=%sm error=%s",
+        host,
+        host_label,
+        window_minutes,
+        (error_msg or '')[:180],
+    )
+    return True
+
+
 def _fallback_111_block_reason(host: str) -> Tuple[bool, str]:
     """Return whether 111 fallback should be skipped for this request.
 
@@ -1086,6 +1166,7 @@ class OllamaService:
 
         # HOTFIX 三主機 retry 鏈（與 generate() 同模式）
         attempted_hosts: List[str] = []
+        skipped_hosts: List[str] = []
         canonical_hosts = _canonical_host_chain()
         allowed_hosts = [
             candidate for candidate in canonical_hosts
@@ -1101,7 +1182,8 @@ class OllamaService:
                     logger.warning("[Embed] 111 fallback disabled; ignoring EMBEDDING_HOST=%s", configured_host)
                 target_host = resolve_ollama_host().rstrip("/")
             if not allow_111_fallback and _is_111_fallback_host(target_host):
-                next_host = next((candidate for candidate in allowed_hosts if candidate not in attempted_hosts), None)
+                visited_hosts = attempted_hosts + skipped_hosts
+                next_host = next((candidate for candidate in allowed_hosts if candidate not in visited_hosts), None)
                 if not next_host:
                     logger.warning("[Embed] 111 fallback disabled; no approved GCP embedding host available")
                     break
@@ -1110,10 +1192,11 @@ class OllamaService:
                     next_host,
                 )
                 target_host = next_host
-            if target_host in attempted_hosts:
+            visited_hosts = attempted_hosts + skipped_hosts
+            if target_host in visited_hosts:
                 next_host = None
                 if target_host in allowed_hosts:
-                    next_host = next((candidate for candidate in allowed_hosts if candidate not in attempted_hosts), None)
+                    next_host = next((candidate for candidate in allowed_hosts if candidate not in visited_hosts), None)
                 if not next_host:
                     break  # cache 還沒過期或同主機，避免無限迴圈
                 logger.info(
@@ -1127,6 +1210,10 @@ class OllamaService:
                 logger.warning("[Embed] skip 111 fallback: %s", block_reason)
                 _clear_resolved_host_cache()
                 break
+            if not allow_111_fallback and _recent_embedding_host_unhealthy(target_host):
+                skipped_hosts.append(target_host)
+                _clear_resolved_host_cache()
+                continue
             attempted_hosts.append(target_host)
 
             vec = _embed_one(target_host)
@@ -1136,11 +1223,12 @@ class OllamaService:
             logger.info(f"[Embed] retry #{attempt+1}/{max_attempts} — {target_host} failed, mark_unhealthy + 取新主機")
 
         if not allow_111_fallback:
-            _open_embedding_gcp_circuit(attempted_hosts)
+            _open_embedding_gcp_circuit(attempted_hosts or skipped_hosts)
             logger.warning(
-                "[Embed] background GCP embedding unavailable; circuit open %ss; tried=%s",
+                "[Embed] background GCP embedding unavailable; circuit open %ss; tried=%s skipped=%s",
                 EMBED_GCP_FAILURE_COOLDOWN_SEC,
                 attempted_hosts,
+                skipped_hosts,
             )
         else:
             logger.error(f"[Embed] all {len(attempted_hosts)} hosts failed; tried={attempted_hosts}")