SPB Git
spboucher.ai

spb/api-ka

Public

API-KA — plateforme centrale : collecte quotidienne des 8 services KA, historisation append-only et API publique sur www.api-ka.com

fastapipostgresqlagregationka

feat: supervision centralisée des connecteurs de l'écosystème KA

- table connector_health (une ligne par service/source, upsert) + job
  src/monitoring/connector_health.py toutes les 2 h dans apika-scheduler
  (lecture seule des GET /api/stats des 8 apps sœurs, timeout 5 s,
  échec d'une app sans impact sur le job)
- classement : broken (≥3 échecs ou 0 résultat ×3), stale (>2× cadence,
  facteur de rotation au niveau source), degraded (<50 % de la médiane),
  ok — cadences par service (lou-ka 1 h … resto-ka 168 h)
- alertes logs/alerts.log uniquement sur transition vers broken/stale
  (anti-spam via l'état précédent mémorisé en table)
- endpoints publics GET /api/v1/monitoring/connectors(/{service}) +
  bloc connectors dans /health
- 15 tests (classement, parsing, anti-spam, endpoints) — suite 48/48

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Simon-Pierre Boucher committed (Aug 18, 2026) parent 02c0fc8 Browse files

8 changed files +1,090 −4

8 changed files
Unified Split
modified src/api/main.py +12 −1
@@ -26,7 +26,17 @@ from fastapi.staticfiles import StaticFiles
26 from src.api import reqstats 26 from src.api import reqstats
27 from src.api.middleware.logging import RequestLoggingMiddleware 27 from src.api.middleware.logging import RequestLoggingMiddleware
28 from src.api.middleware.ratelimit import RateLimitMiddleware 28 from src.api.middleware.ratelimit import RateLimitMiddleware
29−from src.api.routes import agent, auth, iosauth, envelope, health, runs, services, stats 29+from src.api.routes import (
30+ agent,
31+ auth,
32+ envelope,
33+ health,
34+ iosauth,
35+ monitoring,
36+ runs,
37+ services,
38+ stats,
39+)
30 from src.config import SERVICES, get_settings, verify_node 40 from src.config import SERVICES, get_settings, verify_node
31 from src.database.db import init_db 41 from src.database.db import init_db
32 from src.utils.logger import get_logger 42 from src.utils.logger import get_logger
@@ -81,6 +91,7 @@ app.add_middleware(RequestLoggingMiddleware)
81 app.include_router(health.router) 91 app.include_router(health.router)
82 app.include_router(auth.router) 92 app.include_router(auth.router)
83 app.include_router(runs.router) 93 app.include_router(runs.router)
94+app.include_router(monitoring.router)
84 app.include_router(stats.router) 95 app.include_router(stats.router)
85 app.include_router(services.router) 96 app.include_router(services.router)
86 app.include_router(agent.router) 97 app.include_router(agent.router)
modified src/api/routes/health.py +13 −2
@@ -14,13 +14,14 @@ import socket
14 from typing import Any 14 from typing import Any
15 15
16 from fastapi import APIRouter, Depends 16 from fastapi import APIRouter, Depends
17−from sqlalchemy import select 17+from sqlalchemy import func, select
18 from sqlalchemy.orm import Session 18 from sqlalchemy.orm import Session
19 19
20 from src.api.routes import envelope 20 from src.api.routes import envelope
21 from src.config import REQUIRED_NODE, SERVICES 21 from src.config import REQUIRED_NODE, SERVICES
22 from src.database.db import get_db, healthcheck 22 from src.database.db import get_db, healthcheck
23−from src.database.models import CollectionRun 23+from src.database.models import CollectionRun, ConnectorHealth
24+from src.monitoring.connector_health import ALL_STATUSES
24 25
25 router = APIRouter(tags=["health"]) 26 router = APIRouter(tags=["health"])
26 27
@@ -57,6 +58,15 @@ def health(db: Session = Depends(get_db)) -> dict[str, Any]:
57 else None 58 else None
58 ) 59 )
59 60
61+ # Résumé de la supervision des connecteurs de l'écosystème (table
62+ # connector_health, alimentée toutes les 2 h par apika-scheduler).
63+ connectors = dict.fromkeys(ALL_STATUSES, 0)
64+ for status, count in db.execute(
65+ select(ConnectorHealth.status, func.count()).group_by(ConnectorHealth.status)
66+ ):
67+ if status in connectors:
68+ connectors[status] = count
69+
60 data = { 70 data = {
61 "status": "ok" if db_ok else "degraded", 71 "status": "ok" if db_ok else "degraded",
62 "node": hostname, 72 "node": hostname,
@@ -64,5 +74,6 @@ def health(db: Session = Depends(get_db)) -> dict[str, Any]:
64 "node_ok": hostname.split(".")[0].lower() == REQUIRED_NODE, 74 "node_ok": hostname.split(".")[0].lower() == REQUIRED_NODE,
65 "database": "ok" if db_ok else "error", 75 "database": "ok" if db_ok else "error",
66 "last_collections": last_collections, 76 "last_collections": last_collections,
77+ "connectors": connectors,
67 } 78 }
68 return envelope(data) 79 return envelope(data)
added src/api/routes/monitoring.py +112 −0
@@ -0,0 +1,112 @@
1+# ============================================
2+# Projet : API-KA
3+# Fichier : src/api/routes/monitoring.py
4+# Node : m3u96b
5+# Author : Simon-Pierre Boucher
6+# Contact : contact@spboucher.ai
7+# Date : 2026-08-18
8+# ============================================
9+"""Routes publiques de supervision des connecteurs de l'écosystème KA.
10+
11+L'état est produit toutes les 2 h par ``src.monitoring.connector_health``
12+(job apika-scheduler) et stocké dans la table ``connector_health``.
13+"""
14+
15+from __future__ import annotations
16+
17+from typing import Any
18+
19+from fastapi import APIRouter, Depends, HTTPException
20+from sqlalchemy import select
21+from sqlalchemy.orm import Session
22+
23+from src.api.routes import envelope
24+from src.config import SERVICES
25+from src.database.db import get_db
26+from src.database.models import ConnectorHealth
27+from src.monitoring.connector_health import ALL_STATUSES, APP_SOURCE
28+
29+router = APIRouter(prefix="/api/v1/monitoring", tags=["monitoring"])
30+
31+
32+def _row_to_dict(row: ConnectorHealth) -> dict[str, Any]:
33+ return {
34+ "source": row.source,
35+ "status": row.status,
36+ "checked_at": row.checked_at.isoformat() if row.checked_at else None,
37+ "last_success": row.last_success.isoformat() if row.last_success else None,
38+ "found_last": row.found_last,
39+ "median_found": row.median_found,
40+ "consecutive_failures": row.consecutive_failures,
41+ "message": row.message,
42+ }
43+
44+
45+def _empty_counts() -> dict[str, int]:
46+ return dict.fromkeys(ALL_STATUSES, 0)
47+
48+
49+def _group_rows(rows: list[ConnectorHealth]) -> dict[str, Any]:
50+ services: dict[str, Any] = {}
51+ for row in rows:
52+ bucket = services.setdefault(
53+ row.service,
54+ {"summary": _empty_counts(), "app": None, "connectors": []},
55+ )
56+ if row.status in bucket["summary"]:
57+ bucket["summary"][row.status] += 1
58+ payload = _row_to_dict(row)
59+ if row.source == APP_SOURCE:
60+ bucket["app"] = payload
61+ else:
62+ bucket["connectors"].append(payload)
63+ return services
64+
65+
66+@router.get("/connectors")
67+def connectors(db: Session = Depends(get_db)) -> dict[str, Any]:
68+ """État courant de tous les connecteurs, groupé par service."""
69+ rows = (
70+ db.execute(
71+ select(ConnectorHealth).order_by(
72+ ConnectorHealth.service, ConnectorHealth.source
73+ )
74+ )
75+ .scalars()
76+ .all()
77+ )
78+ services = _group_rows(rows)
79+
80+ summary = _empty_counts()
81+ for bucket in services.values():
82+ for status, count in bucket["summary"].items():
83+ summary[status] += count
84+
85+ data = {
86+ "summary": summary,
87+ "services": services,
88+ }
89+ return envelope(data, extra_meta={"services_total": len(services)})
90+
91+
92+@router.get("/connectors/{service}")
93+def connectors_service(
94+ service: str, db: Session = Depends(get_db)
95+) -> dict[str, Any]:
96+ """État courant des connecteurs d'un service donné."""
97+ if service not in SERVICES:
98+ raise HTTPException(status_code=404, detail=f"Service inconnu : {service}")
99+ rows = (
100+ db.execute(
101+ select(ConnectorHealth)
102+ .where(ConnectorHealth.service == service)
103+ .order_by(ConnectorHealth.source)
104+ )
105+ .scalars()
106+ .all()
107+ )
108+ bucket = _group_rows(rows).get(
109+ service, {"summary": _empty_counts(), "app": None, "connectors": []}
110+ )
111+ data = {"service": service, **bucket}
112+ return envelope(data)
modified src/database/models.py +38 −0
@@ -181,3 +181,41 @@ class ApiRequest(Base):
181 endpoint: Mapped[str] = mapped_column(Text, nullable=False) 181 endpoint: Mapped[str] = mapped_column(Text, nullable=False)
182 status: Mapped[int] = mapped_column(Integer, nullable=False) 182 status: Mapped[int] = mapped_column(Integer, nullable=False)
183 duration_ms: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) 183 duration_ms: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
184+
185+
186+class ConnectorHealth(Base):
187+ """État courant de chaque connecteur de l'écosystème KA (supervision).
188+
189+ Une ligne par couple (service, source), upsertée toutes les 2 h par
190+ ``src.monitoring.connector_health``. La source virtuelle ``_app``
191+ représente l'état global de l'app. La ligne mémorise aussi l'état
192+ précédent (anti-spam des alertes : on n'alerte qu'à la transition).
193+ """
194+
195+ __tablename__ = "connector_health"
196+ __table_args__ = (
197+ UniqueConstraint(
198+ "service", "source", name="uq_connector_health_service_source"
199+ ),
200+ Index("ix_connector_health_service", "service"),
201+ Index("ix_connector_health_status", "status"),
202+ )
203+
204+ id: Mapped[int] = mapped_column(BigIntPK, primary_key=True, autoincrement=True)
205+ service: Mapped[str] = mapped_column(Text, nullable=False)
206+ source: Mapped[str] = mapped_column(Text, nullable=False)
207+ checked_at: Mapped[datetime.datetime] = mapped_column(
208+ DateTime(timezone=True), nullable=False, server_default=func.now()
209+ )
210+ status: Mapped[str] = mapped_column(
211+ Text, nullable=False
212+ ) # ok / degraded / broken / stale
213+ last_success: Mapped[datetime.datetime | None] = mapped_column(
214+ DateTime(timezone=True), nullable=True
215+ )
216+ found_last: Mapped[int | None] = mapped_column(Integer, nullable=True)
217+ median_found: Mapped[float | None] = mapped_column(Float, nullable=True)
218+ consecutive_failures: Mapped[int] = mapped_column(
219+ Integer, nullable=False, default=0
220+ )
221+ message: Mapped[str | None] = mapped_column(Text, nullable=True)
added src/monitoring/__init__.py +9 −0
@@ -0,0 +1,9 @@
1+# ============================================
2+# Projet : API-KA
3+# Fichier : src/monitoring/__init__.py
4+# Node : m3u96b
5+# Author : Simon-Pierre Boucher
6+# Contact : contact@spboucher.ai
7+# Date : 2026-08-18
8+# ============================================
9+"""Supervision centralisée des connecteurs de l'écosystème KA."""
added src/monitoring/connector_health.py +523 −0
@@ -0,0 +1,523 @@
1+# ============================================
2+# Projet : API-KA
3+# Fichier : src/monitoring/connector_health.py
4+# Node : m3u96b
5+# Author : Simon-Pierre Boucher
6+# Contact : contact@spboucher.ai
7+# Date : 2026-08-18
8+# ============================================
9+"""Supervision centralisée des connecteurs des 8 apps de l'écosystème KA.
10+
11+Toutes les 2 h, le job lit le ``GET /api/stats`` de chaque app sœur (LAN,
12+lecture seule, timeout 5 s) et en déduit un état par source quand le détail
13+existe (``recent_syncs`` / ``sync_log``), plus un état global par app
14+(source virtuelle ``_app``). L'échec d'une app n'interrompt jamais le job.
15+
16+Règles de classement :
17+
18+- ``broken`` : ≥ 3 synchros consécutives en échec ou à 0 résultat
19+ (ou app injoignable 3 fois de suite) ;
20+- ``stale`` : aucun sync réussi depuis > 2× la cadence attendue du service
21+ (au niveau source, la cadence est multipliée par un facteur de
22+ rotation : les apps synchronisent leurs sources par lots, une
23+ source individuelle repasse donc moins souvent que l'app) ;
24+- ``degraded`` : dernier volume < 50 % de la médiane observée, dernier sync en
25+ échec (< 3), ou app injoignable (< 3) ;
26+- ``ok`` : sinon.
27+
28+Anti-spam : l'état précédent est mémorisé dans la table ``connector_health``
29+(une ligne par couple service/source, upsert à chaque run) ; une alerte
30+n'est écrite dans ``logs/alerts.log`` que lors d'une TRANSITION vers
31+``broken`` ou ``stale``.
32+"""
33+
34+from __future__ import annotations
35+
36+import datetime
37+import math
38+import statistics
39+import time
40+from dataclasses import dataclass, field
41+from typing import Any
42+from urllib.parse import urlsplit
43+
44+import httpx
45+from sqlalchemy import select
46+
47+from src.config import SERVICES, get_settings
48+from src.database.db import session_scope
49+from src.database.models import ConnectorHealth
50+from src.utils.logger import alert, get_logger
51+
52+# Source virtuelle représentant l'état global d'une app.
53+APP_SOURCE = "_app"
54+
55+STATS_TIMEOUT_SECONDS = 5.0
56+
57+STATUS_OK = "ok"
58+STATUS_DEGRADED = "degraded"
59+STATUS_BROKEN = "broken"
60+STATUS_STALE = "stale"
61+ALL_STATUSES = (STATUS_OK, STATUS_DEGRADED, STATUS_BROKEN, STATUS_STALE)
62+
63+# broken après N échecs (ou 0 résultat) consécutifs.
64+BROKEN_THRESHOLD = 3
65+# degraded si volume < 50 % de la médiane.
66+DEGRADED_RATIO = 0.5
67+# stale si aucun succès depuis > 2× la cadence attendue.
68+STALE_FACTOR = 2
69+
70+# Cadence de sync attendue par service (heures).
71+EXPECTED_CADENCE_HOURS: dict[str, float] = {
72+ "louka": 1,
73+ "immoka": 4,
74+ "autoka": 2,
75+ "foodka": 6,
76+ "fabrika": 6,
77+ "sortika": 1,
78+ "creaka": 24,
79+ "restoka": 168,
80+}
81+
82+
83+@dataclass
84+class SyncEntry:
85+ """Une entrée de journal de sync exposée par une app sœur."""
86+
87+ ts: float
88+ found: int
89+ ok: bool
90+ message: str = ""
91+
92+
93+@dataclass
94+class SourceState:
95+ """État précédent d'un connecteur (relu depuis ``connector_health``)."""
96+
97+ status: str | None = None
98+ last_success: datetime.datetime | None = None
99+ found_last: int | None = None
100+ median_found: float | None = None
101+ consecutive_failures: int = 0
102+ checked_at: datetime.datetime | None = None
103+
104+
105+@dataclass
106+class Assessment:
107+ """Résultat du classement d'un connecteur pour le run courant."""
108+
109+ status: str
110+ last_success: datetime.datetime | None
111+ found_last: int | None
112+ median_found: float | None
113+ consecutive_failures: int
114+ message: str = ""
115+ entries: list[SyncEntry] = field(default_factory=list)
116+
117+
118+def _utc(ts: float) -> datetime.datetime:
119+ return datetime.datetime.fromtimestamp(ts, tz=datetime.UTC)
120+
121+
122+def assess_source(
123+ entries: list[SyncEntry],
124+ prev: SourceState,
125+ now: float,
126+ stale_after_seconds: float,
127+) -> Assessment:
128+ """Classe un connecteur : broken > stale > degraded > ok.
129+
130+ Fonction pure (testable) : combine les entrées de la fenêtre courante
131+ avec l'état précédent (streak d'échecs, dernier succès, médiane glissante).
132+ """
133+ entries = sorted(entries, key=lambda e: e.ts, reverse=True)
134+
135+ # Streak d'échecs : seules les entrées jamais vues (postérieures au dernier
136+ # passage) alimentent le compteur, du plus ancien au plus récent.
137+ prev_checked_ts = prev.checked_at.timestamp() if prev.checked_at else None
138+ new_entries = [
139+ e for e in entries if prev_checked_ts is None or e.ts > prev_checked_ts
140+ ]
141+ consecutive_failures = prev.consecutive_failures
142+ for entry in reversed(new_entries):
143+ if not entry.ok or entry.found <= 0:
144+ consecutive_failures += 1
145+ else:
146+ consecutive_failures = 0
147+
148+ # Dernier sync réussi (ok ET au moins un résultat).
149+ last_success = prev.last_success
150+ for entry in entries:
151+ if entry.ok and entry.found > 0:
152+ candidate = _utc(entry.ts)
153+ if last_success is None or candidate > last_success:
154+ last_success = candidate
155+ break
156+
157+ # Médiane glissante du volume (entrées ok de la fenêtre + médiane passée).
158+ ok_founds = [float(e.found) for e in entries if e.ok]
159+ values = ok_founds + (
160+ [float(prev.median_found)] if prev.median_found is not None else []
161+ )
162+ median_found = statistics.median(values) if values else None
163+
164+ found_last = entries[0].found if entries else prev.found_last
165+
166+ last_message = next((e.message for e in entries if e.message), "")
167+
168+ if consecutive_failures >= BROKEN_THRESHOLD:
169+ message = (
170+ f"{consecutive_failures} synchros consécutives en échec ou à 0 résultat"
171+ )
172+ if last_message and last_message != "ok":
173+ message += f" (dernier message : {last_message})"
174+ status = STATUS_BROKEN
175+ elif last_success is not None and (now - last_success.timestamp()) > stale_after_seconds:
176+ hours = (now - last_success.timestamp()) / 3600
177+ message = (
178+ f"aucun sync réussi depuis {hours:.1f} h "
179+ f"(seuil : {stale_after_seconds / 3600:.0f} h)"
180+ )
181+ status = STATUS_STALE
182+ elif last_success is None:
183+ status = STATUS_DEGRADED
184+ message = "aucun sync réussi observé pour l'instant"
185+ elif entries and not entries[0].ok:
186+ status = STATUS_DEGRADED
187+ message = (
188+ f"dernier sync en échec ({consecutive_failures}/{BROKEN_THRESHOLD})"
189+ )
190+ if last_message and last_message != "ok":
191+ message += f" — {last_message}"
192+ elif (
193+ found_last is not None
194+ and median_found
195+ and found_last < DEGRADED_RATIO * median_found
196+ ):
197+ status = STATUS_DEGRADED
198+ message = (
199+ f"volume {found_last} < 50 % de la médiane ({median_found:.0f})"
200+ )
201+ else:
202+ status = STATUS_OK
203+ message = "ok"
204+
205+ return Assessment(
206+ status=status,
207+ last_success=last_success,
208+ found_last=found_last,
209+ median_found=median_found,
210+ consecutive_failures=consecutive_failures,
211+ message=message,
212+ entries=entries,
213+ )
214+
215+
216+def assess_unreachable(prev: SourceState, error: str) -> Assessment:
217+ """Classe l'état global d'une app dont le /api/stats est injoignable."""
218+ consecutive_failures = prev.consecutive_failures + 1
219+ status = (
220+ STATUS_BROKEN if consecutive_failures >= BROKEN_THRESHOLD else STATUS_DEGRADED
221+ )
222+ return Assessment(
223+ status=status,
224+ last_success=prev.last_success,
225+ found_last=prev.found_last,
226+ median_found=prev.median_found,
227+ consecutive_failures=consecutive_failures,
228+ message=f"GET /api/stats injoignable ({error})",
229+ )
230+
231+
232+def parse_sync_entries(payload: dict[str, Any]) -> dict[str, list[SyncEntry]]:
233+ """Extrait les entrées de sync par source depuis un payload /api/stats.
234+
235+ Comprend les deux formats observés dans l'écosystème :
236+ ``recent_syncs`` (lou-ka, immo-ka, auto-ka, food-ka, resto-ka) et
237+ ``sync_log`` par magasin (fabri-ka, clé ``store_id`` + ``status``).
238+ """
239+ raw = payload.get("recent_syncs") or payload.get("sync_log") or []
240+ by_source: dict[str, list[SyncEntry]] = {}
241+ for item in raw:
242+ if not isinstance(item, dict):
243+ continue
244+ source = item.get("source") or item.get("store_id")
245+ if not source:
246+ continue
247+ if "ok" in item:
248+ ok = bool(item.get("ok"))
249+ else:
250+ ok = str(item.get("status", "ok")).lower() == "ok"
251+ try:
252+ entry = SyncEntry(
253+ ts=float(item.get("ts") or 0),
254+ found=int(item.get("found") or 0),
255+ ok=ok,
256+ message=str(item.get("message") or ""),
257+ )
258+ except (TypeError, ValueError):
259+ continue
260+ by_source.setdefault(str(source), []).append(entry)
261+ return by_source
262+
263+
264+_GLOBAL_TOTAL_KEYS = ("total_active", "total", "creators", "restaurants")
265+
266+
267+def build_app_entry(
268+ service: str,
269+ payload: dict[str, Any],
270+ by_source: dict[str, list[SyncEntry]],
271+ now: float,
272+) -> SyncEntry:
273+ """Synthétise UNE entrée représentant l'état global de l'app.
274+
275+ - apps avec journal de sync : agrégat de la fenêtre (ts le plus récent,
276+ somme des volumes ok) ;
277+ - crea-ka : ``last_sync`` global + total ``creators`` ;
278+ - sorti-ka (aucun horodatage exposé) : joignabilité + volume total.
279+ """
280+ all_entries = [e for lst in by_source.values() for e in lst]
281+ if all_entries:
282+ return SyncEntry(
283+ ts=max(e.ts for e in all_entries),
284+ found=sum(e.found for e in all_entries if e.ok),
285+ ok=any(e.ok for e in all_entries),
286+ )
287+
288+ last_sync = payload.get("last_sync")
289+ if isinstance(last_sync, str) and last_sync:
290+ try:
291+ ts = datetime.datetime.fromisoformat(
292+ last_sync.replace("Z", "+00:00")
293+ ).timestamp()
294+ except ValueError:
295+ ts = now
296+ total = _global_total(payload)
297+ return SyncEntry(ts=ts, found=total, ok=True, message="last_sync global")
298+
299+ # Fallback (ex. sorti-ka) : l'app ne publie aucun horodatage de sync.
300+ return SyncEntry(
301+ ts=now,
302+ found=_global_total(payload),
303+ ok=True,
304+ message="pas d'horodatage de sync exposé — joignabilité + volume total",
305+ )
306+
307+
308+def _global_total(payload: dict[str, Any]) -> int:
309+ for key in _GLOBAL_TOTAL_KEYS:
310+ value = payload.get(key)
311+ if isinstance(value, (int, float)):
312+ return int(value)
313+ totals = payload.get("totals")
314+ if isinstance(totals, dict):
315+ for value in totals.values():
316+ if isinstance(value, (int, float)):
317+ return int(value)
318+ return 0
319+
320+
321+def rotation_factor(
322+ payload: dict[str, Any], by_source: dict[str, list[SyncEntry]]
323+) -> int:
324+ """Facteur de rotation des sources : les apps synchronisent leurs sources
325+ par lots (fenêtre ~20 entrées pour parfois >100 sources). Une source
326+ individuelle repasse donc toutes les ``rotation × cadence`` heures environ ;
327+ le seuil de staleness par source en tient compte pour ne pas générer de
328+ fausses alertes de masse."""
329+ window = sum(len(v) for v in by_source.values())
330+ hint = payload.get("sources")
331+ if not isinstance(hint, (int, float)):
332+ totals = payload.get("totals")
333+ hint = totals.get("stores_live") if isinstance(totals, dict) else 0
334+ total = max(int(hint or 0), len(by_source))
335+ if window <= 0 or total <= 0:
336+ return 1
337+ return max(1, math.ceil(total / window))
338+
339+
340+def _stats_url(service: str) -> str | None:
341+ """Dérive l'URL /api/stats de l'app depuis la SOURCE_URL du .env."""
342+ source_url = get_settings().source_urls.get(service, "")
343+ if not source_url:
344+ return None
345+ parts = urlsplit(source_url)
346+ if not parts.scheme or not parts.netloc:
347+ return None
348+ return f"{parts.scheme}://{parts.netloc}/api/stats"
349+
350+
351+def _fetch_stats(url: str) -> dict[str, Any]:
352+ """GET /api/stats (lecture seule, timeout court)."""
353+ response = httpx.get(url, timeout=STATS_TIMEOUT_SECONDS, follow_redirects=True)
354+ response.raise_for_status()
355+ payload = response.json()
356+ if not isinstance(payload, dict):
357+ raise ValueError("payload /api/stats inattendu (dict requis)")
358+ return payload
359+
360+
361+def _state_from_row(row: ConnectorHealth | None) -> SourceState:
362+ if row is None:
363+ return SourceState()
364+ return SourceState(
365+ status=row.status,
366+ last_success=_ensure_utc(row.last_success),
367+ found_last=row.found_last,
368+ median_found=row.median_found,
369+ consecutive_failures=row.consecutive_failures or 0,
370+ checked_at=_ensure_utc(row.checked_at),
371+ )
372+
373+
374+def _ensure_utc(value: datetime.datetime | None) -> datetime.datetime | None:
375+ if value is None:
376+ return None
377+ if value.tzinfo is None:
378+ return value.replace(tzinfo=datetime.UTC)
379+ return value
380+
381+
382+def _apply(
383+ session: Any,
384+ rows: dict[str, ConnectorHealth],
385+ service: str,
386+ source: str,
387+ assessment: Assessment,
388+ now: float,
389+) -> str:
390+ """Upsert de la ligne connector_health + alerte sur transition broken/stale."""
391+ row = rows.get(source)
392+ previous_status = row.status if row is not None else None
393+
394+ checked_at = _utc(now)
395+ if row is None:
396+ row = ConnectorHealth(service=service, source=source)
397+ session.add(row)
398+ rows[source] = row
399+ row.checked_at = checked_at
400+ row.status = assessment.status
401+ row.last_success = assessment.last_success
402+ row.found_last = assessment.found_last
403+ row.median_found = assessment.median_found
404+ row.consecutive_failures = assessment.consecutive_failures
405+ row.message = assessment.message
406+
407+ if assessment.status in (STATUS_BROKEN, STATUS_STALE) and (
408+ previous_status != assessment.status
409+ ):
410+ alert(
411+ f"[connecteurs] {service}/{source} : {assessment.status} — "
412+ f"{assessment.message}"
413+ )
414+ return assessment.status
415+
416+
417+def _check_service(service: str, now: float) -> dict[str, Any]:
418+ """Collecte et classe tous les connecteurs d'un service. Ne lève jamais
419+ pour cause d'app injoignable (l'échec est un état, pas une exception)."""
420+ url = _stats_url(service)
421+ if url is None:
422+ return {"service": service, "skipped": "aucune SOURCE_URL configurée"}
423+
424+ cadence_seconds = EXPECTED_CADENCE_HOURS.get(service, 24) * 3600
425+ app_stale_after = STALE_FACTOR * cadence_seconds
426+
427+ counts = dict.fromkeys(ALL_STATUSES, 0)
428+
429+ try:
430+ payload = _fetch_stats(url)
431+ except Exception as exc: # httpx, JSON, payload inattendu…
432+ with session_scope() as session:
433+ rows = _load_rows(session, service)
434+ assessment = assess_unreachable(
435+ _state_from_row(rows.get(APP_SOURCE)), str(exc)
436+ )
437+ counts[_apply(session, rows, service, APP_SOURCE, assessment, now)] += 1
438+ return {"service": service, "url": url, "reachable": False, "counts": counts}
439+
440+ by_source = parse_sync_entries(payload)
441+ app_entry = build_app_entry(service, payload, by_source, now)
442+ source_stale_after = app_stale_after * rotation_factor(payload, by_source)
443+
444+ with session_scope() as session:
445+ rows = _load_rows(session, service)
446+
447+ assessment = assess_source(
448+ [app_entry], _state_from_row(rows.get(APP_SOURCE)), now, app_stale_after
449+ )
450+ counts[_apply(session, rows, service, APP_SOURCE, assessment, now)] += 1
451+
452+ for source, entries in by_source.items():
453+ assessment = assess_source(
454+ entries, _state_from_row(rows.get(source)), now, source_stale_after
455+ )
456+ counts[_apply(session, rows, service, source, assessment, now)] += 1
457+
458+ # Sources connues mais absentes de la fenêtre courante : réévaluer la
459+ # staleness (le streak et la médiane restent inchangés).
460+ for source in list(rows):
461+ if source == APP_SOURCE or source in by_source:
462+ continue
463+ assessment = assess_source(
464+ [], _state_from_row(rows.get(source)), now, source_stale_after
465+ )
466+ counts[_apply(session, rows, service, source, assessment, now)] += 1
467+
468+ return {
469+ "service": service,
470+ "url": url,
471+ "reachable": True,
472+ "sources_in_window": len(by_source),
473+ "counts": counts,
474+ }
475+
476+
477+def _load_rows(session: Any, service: str) -> dict[str, ConnectorHealth]:
478+ return {
479+ row.source: row
480+ for row in session.execute(
481+ select(ConnectorHealth).where(ConnectorHealth.service == service)
482+ ).scalars()
483+ }
484+
485+
486+def run_connector_health_check(now: float | None = None) -> dict[str, Any]:
487+ """Point d'entrée du job (toutes les 2 h) : vérifie les 8 apps.
488+
489+ L'échec d'une app (réseau, payload) ne fait jamais échouer le job.
490+ """
491+ logger = get_logger("apika.monitoring")
492+ now = now if now is not None else time.time()
493+ results: list[dict[str, Any]] = []
494+ for service in SERVICES:
495+ try:
496+ results.append(_check_service(service, now))
497+ except Exception as exc: # filet de sécurité (ex. DB) — jamais de crash
498+ logger.error(
499+ "Échec de la supervision d'un service",
500+ extra={"service": service, "error": str(exc)},
501+ )
502+ results.append({"service": service, "error": str(exc)})
503+
504+ summary = {
505+ "checked_at": _utc(now).isoformat(),
506+ "services": results,
507+ }
508+ logger.info("Supervision des connecteurs terminée", extra=summary)
509+ return summary
510+
511+
512+def main() -> None:
513+ """Exécution manuelle : ``python -m src.monitoring.connector_health``."""
514+ import json
515+
516+ from src.database.db import init_db
517+
518+ init_db()
519+ print(json.dumps(run_connector_health_check(), ensure_ascii=False, indent=2))
520+
521+
522+if __name__ == "__main__":
523+ main()
modified src/scheduler/daily_job.py +19 −1
@@ -23,10 +23,12 @@ from typing import Any
23 23
24 from apscheduler.schedulers.blocking import BlockingScheduler 24 from apscheduler.schedulers.blocking import BlockingScheduler
25 from apscheduler.triggers.cron import CronTrigger 25 from apscheduler.triggers.cron import CronTrigger
26+from apscheduler.triggers.interval import IntervalTrigger
26 27
27 from src.collectors import build_collectors 28 from src.collectors import build_collectors
28 from src.config import get_settings, verify_node 29 from src.config import get_settings, verify_node
29 from src.database.db import init_db 30 from src.database.db import init_db
31+from src.monitoring.connector_health import run_connector_health_check
30 from src.scheduler.backfill import run_backfill 32 from src.scheduler.backfill import run_backfill
31 from src.utils.logger import alert, get_logger 33 from src.utils.logger import alert, get_logger
32 34
@@ -129,9 +131,25 @@ def main() -> None:
129 coalesce=True, 131 coalesce=True,
130 misfire_grace_time=3600, 132 misfire_grace_time=3600,
131 ) 133 )
134+ # Supervision des connecteurs de l'écosystème (lecture seule des
135+ # /api/stats des 8 apps sœurs) : toutes les 2 h, premier passage 90 s
136+ # après le démarrage du scheduler.
137+ scheduler.add_job(
138+ run_connector_health_check,
139+ IntervalTrigger(hours=2),
140+ id="apika_connector_health",
141+ max_instances=1,
142+ coalesce=True,
143+ misfire_grace_time=900,
144+ next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=90),
145+ )
132 logger.info( 146 logger.info(
133 "Scheduler démarré", 147 "Scheduler démarré",
134− extra={"daily_run_hour": settings.daily_run_hour, "node": settings.node_name}, 148+ extra={
149+ "daily_run_hour": settings.daily_run_hour,
150+ "connector_health_interval_hours": 2,
151+ "node": settings.node_name,
152+ },
135 ) 153 )
136 scheduler.start() 154 scheduler.start()
137 155
added tests/test_connector_health.py +364 −0
@@ -0,0 +1,364 @@
1+# ============================================
2+# Projet : API-KA
3+# Fichier : tests/test_connector_health.py
4+# Node : m3u96b
5+# Author : Simon-Pierre Boucher
6+# Contact : contact@spboucher.ai
7+# Date : 2026-08-18
8+# ============================================
9+"""Tests de la supervision des connecteurs : classement d'état (fonctions pures),
10+parsing des /api/stats, upsert + anti-spam des alertes, endpoints publics."""
11+
12+from __future__ import annotations
13+
14+import datetime
15+
16+import pytest
17+from fastapi.testclient import TestClient
18+from sqlalchemy import delete, select
19+
20+from src.api.main import app
21+from src.database.db import session_scope
22+from src.database.models import ConnectorHealth
23+from src.monitoring import connector_health as ch
24+from src.monitoring.connector_health import (
25+ APP_SOURCE,
26+ SourceState,
27+ SyncEntry,
28+ assess_source,
29+ assess_unreachable,
30+ build_app_entry,
31+ parse_sync_entries,
32+ rotation_factor,
33+ run_connector_health_check,
34+)
35+
36+NOW = 1_787_040_000.0
37+HOUR = 3600.0
38+
39+
40+def _entry(hours_ago: float, found: int = 100, ok: bool = True, message: str = "") -> SyncEntry:
41+ return SyncEntry(ts=NOW - hours_ago * HOUR, found=found, ok=ok, message=message)
42+
43+
44+# ---------------------------------------------------------------- classement
45+
46+
47+def test_assess_ok_nominal():
48+ result = assess_source(
49+ [_entry(0.5, found=100), _entry(1.5, found=110)],
50+ SourceState(),
51+ NOW,
52+ stale_after_seconds=4 * HOUR,
53+ )
54+ assert result.status == "ok"
55+ assert result.found_last == 100
56+ assert result.consecutive_failures == 0
57+ assert result.last_success is not None
58+
59+
60+def test_assess_broken_apres_trois_echecs():
61+ entries = [
62+ _entry(0.5, found=0, ok=False, message="timeout"),
63+ _entry(1.5, found=0, ok=False),
64+ _entry(2.5, found=0, ok=False),
65+ ]
66+ result = assess_source(entries, SourceState(), NOW, stale_after_seconds=100 * HOUR)
67+ assert result.status == "broken"
68+ assert result.consecutive_failures == 3
69+
70+
71+def test_assess_broken_apres_trois_zero_resultats():
72+ # Syncs "réussis" mais vides ×3 → broken.
73+ entries = [_entry(h, found=0, ok=True) for h in (0.5, 1.5, 2.5)]
74+ result = assess_source(entries, SourceState(), NOW, stale_after_seconds=100 * HOUR)
75+ assert result.status == "broken"
76+ assert result.consecutive_failures == 3
77+
78+
79+def test_assess_streak_cumule_avec_etat_precedent():
80+ prev = SourceState(
81+ status="degraded",
82+ consecutive_failures=2,
83+ checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),
84+ )
85+ # Une seule nouvelle entrée en échec suffit alors pour atteindre 3.
86+ result = assess_source(
87+ [_entry(0.5, found=0, ok=False)], prev, NOW, stale_after_seconds=100 * HOUR
88+ )
89+ assert result.status == "broken"
90+ assert result.consecutive_failures == 3
91+
92+
93+def test_assess_succes_remet_le_streak_a_zero():
94+ prev = SourceState(
95+ consecutive_failures=2,
96+ checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),
97+ )
98+ result = assess_source(
99+ [_entry(0.5, found=80)], prev, NOW, stale_after_seconds=100 * HOUR
100+ )
101+ assert result.status == "ok"
102+ assert result.consecutive_failures == 0
103+
104+
105+def test_assess_stale_sans_sync_recent():
106+ prev = SourceState(
107+ status="ok",
108+ last_success=datetime.datetime.fromtimestamp(NOW - 10 * HOUR, tz=datetime.UTC),
109+ checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),
110+ median_found=100.0,
111+ found_last=100,
112+ )
113+ # Cadence 1 h → stale au-delà de 2 h sans succès.
114+ result = assess_source([], prev, NOW, stale_after_seconds=2 * HOUR)
115+ assert result.status == "stale"
116+
117+
118+def test_assess_degraded_si_volume_sous_la_moitie_de_la_mediane():
119+ prev = SourceState(
120+ median_found=200.0,
121+ last_success=datetime.datetime.fromtimestamp(NOW - 1 * HOUR, tz=datetime.UTC),
122+ )
123+ result = assess_source(
124+ [_entry(0.5, found=40)], prev, NOW, stale_after_seconds=100 * HOUR
125+ )
126+ assert result.status == "degraded"
127+ assert "50 %" in result.message
128+
129+
130+def test_assess_priorite_broken_avant_stale():
131+ prev = SourceState(
132+ last_success=datetime.datetime.fromtimestamp(NOW - 50 * HOUR, tz=datetime.UTC),
133+ consecutive_failures=5,
134+ checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),
135+ )
136+ result = assess_source([], prev, NOW, stale_after_seconds=2 * HOUR)
137+ assert result.status == "broken"
138+
139+
140+def test_assess_unreachable_degrade_puis_broken():
141+ first = assess_unreachable(SourceState(), "connexion refusée")
142+ assert first.status == "degraded"
143+ assert first.consecutive_failures == 1
144+ third = assess_unreachable(
145+ SourceState(consecutive_failures=2), "connexion refusée"
146+ )
147+ assert third.status == "broken"
148+
149+
150+# ------------------------------------------------------------------- parsing
151+
152+
153+def test_parse_recent_syncs_et_sync_log():
154+ payload = {
155+ "recent_syncs": [
156+ {"source": "kangalou", "ts": NOW - 100, "found": 7916, "ok": 1, "message": "ok"},
157+ {"source": "kangalou", "ts": NOW - 4000, "found": 0, "ok": 0, "message": "err"},
158+ ]
159+ }
160+ by_source = parse_sync_entries(payload)
161+ assert list(by_source) == ["kangalou"]
162+ assert len(by_source["kangalou"]) == 2
163+ assert by_source["kangalou"][1].ok is False
164+
165+ fabrika = {
166+ "sync_log": [
167+ {"ts": NOW - 50, "store_id": "daigneau.ca", "found": 25, "status": "ok"},
168+ {"ts": NOW - 60, "store_id": "x.com", "found": 0, "status": "error"},
169+ ]
170+ }
171+ by_store = parse_sync_entries(fabrika)
172+ assert by_store["daigneau.ca"][0].ok is True
173+ assert by_store["x.com"][0].ok is False
174+
175+
176+def test_build_app_entry_last_sync_et_fallback():
177+ # crea-ka : last_sync global ISO.
178+ creaka = {"creators": 4882, "last_sync": "2026-08-18T06:27:51Z"}
179+ entry = build_app_entry("creaka", creaka, {}, NOW)
180+ assert entry.ok is True
181+ assert entry.found == 4882
182+ assert entry.ts != NOW
183+
184+ # sorti-ka : aucun horodatage → joignabilité + volume total.
185+ sortika = {"total_active": 15820, "sources": 10}
186+ entry = build_app_entry("sortika", sortika, {}, NOW)
187+ assert entry.ts == NOW
188+ assert entry.found == 15820
189+
190+
191+def test_rotation_factor():
192+ by_source = {f"s{i}": [_entry(1)] for i in range(20)}
193+ assert rotation_factor({"sources": 206}, by_source) == 11
194+ assert rotation_factor({"sources": 3}, by_source) == 1
195+ assert rotation_factor({}, {}) == 1
196+
197+
198+# ------------------------------------------- job complet : upsert + anti-spam
199+
200+
201+@pytest.fixture()
202+def clean_connector_table():
203+ with session_scope() as session:
204+ session.execute(delete(ConnectorHealth))
205+ yield
206+ with session_scope() as session:
207+ session.execute(delete(ConnectorHealth))
208+
209+
210+def _fake_louka_payload(ok: bool) -> dict:
211+ return {
212+ "total": 44343,
213+ "sources": 2,
214+ "recent_syncs": [
215+ {
216+ "source": "kangalou",
217+ "ts": NOW - 600,
218+ "found": 0 if not ok else 7916,
219+ "ok": 0 if not ok else 1,
220+ "message": "err" if not ok else "ok",
221+ }
222+ ],
223+ }
224+
225+
226+def test_job_upsert_et_alerte_sans_repetition(
227+ monkeypatch: pytest.MonkeyPatch, clean_connector_table, isolated_logs_dir
228+):
229+ """3 runs en échec → broken + UNE alerte ; run suivant identique → pas de
230+ nouvelle alerte (anti-spam) ; retour au succès → ok."""
231+ alerts_file = isolated_logs_dir / "alerts.log"
232+ baseline = alerts_file.read_text() if alerts_file.exists() else ""
233+
234+ monkeypatch.setattr(ch, "_stats_url", lambda service: "http://test/api/stats")
235+ payload = {"value": _fake_louka_payload(ok=False)}
236+ monkeypatch.setattr(ch, "_fetch_stats", lambda url: payload["value"])
237+
238+ # Trois runs espacés de 2 h, chacun voyant UNE nouvelle synchro en échec
239+ # (une entrée déjà comptée — ts <= checked_at précédent — ne recompte pas).
240+ for i in range(3):
241+ run_at = NOW + i * 2 * HOUR
242+ p = _fake_louka_payload(ok=False)
243+ p["recent_syncs"][0]["ts"] = run_at - 600
244+ payload["value"] = p
245+ ch._check_service("louka", run_at)
246+
247+ with session_scope() as session:
248+ row = session.execute(
249+ select(ConnectorHealth).where(
250+ ConnectorHealth.service == "louka",
251+ ConnectorHealth.source == "kangalou",
252+ )
253+ ).scalar_one()
254+ assert row.status == "broken"
255+ assert row.consecutive_failures == 3
256+
257+ content = alerts_file.read_text()[len(baseline):]
258+ assert content.count("[connecteurs] louka/kangalou : broken") == 1
259+
260+ # Même état au run suivant → aucune nouvelle alerte.
261+ run_at = NOW + 3 * 2 * HOUR
262+ p = _fake_louka_payload(ok=False)
263+ p["recent_syncs"][0]["ts"] = run_at - 600
264+ payload["value"] = p
265+ ch._check_service("louka", run_at)
266+ content = alerts_file.read_text()[len(baseline):]
267+ assert content.count("[connecteurs] louka/kangalou : broken") == 1
268+
269+ # Retour au succès → ok, streak remis à zéro.
270+ run_at = NOW + 4 * 2 * HOUR
271+ p = _fake_louka_payload(ok=True)
272+ p["recent_syncs"][0]["ts"] = run_at - 600
273+ payload["value"] = p
274+ ch._check_service("louka", run_at)
275+ with session_scope() as session:
276+ row = session.execute(
277+ select(ConnectorHealth).where(
278+ ConnectorHealth.service == "louka",
279+ ConnectorHealth.source == "kangalou",
280+ )
281+ ).scalar_one()
282+ assert row.status == "ok"
283+ assert row.consecutive_failures == 0
284+
285+
286+def test_job_app_injoignable_ne_crashe_pas(
287+ monkeypatch: pytest.MonkeyPatch, clean_connector_table
288+):
289+ monkeypatch.setattr(ch, "_stats_url", lambda service: "http://test/api/stats")
290+
291+ def _boom(url):
292+ raise ConnectionError("refusée")
293+
294+ monkeypatch.setattr(ch, "_fetch_stats", _boom)
295+ summary = run_connector_health_check(NOW)
296+ assert len(summary["services"]) == 8
297+ with session_scope() as session:
298+ row = session.execute(
299+ select(ConnectorHealth).where(
300+ ConnectorHealth.service == "louka",
301+ ConnectorHealth.source == APP_SOURCE,
302+ )
303+ ).scalar_one()
304+ assert row.status == "degraded"
305+ assert row.consecutive_failures == 1
306+
307+
308+# ----------------------------------------------------------------- endpoints
309+
310+
311+def test_endpoints_monitoring(clean_connector_table):
312+ now = datetime.datetime.now(tz=datetime.UTC)
313+ with session_scope() as session:
314+ session.add(
315+ ConnectorHealth(
316+ service="louka",
317+ source=APP_SOURCE,
318+ checked_at=now,
319+ status="ok",
320+ last_success=now,
321+ found_last=44343,
322+ median_found=44000.0,
323+ consecutive_failures=0,
324+ message="ok",
325+ )
326+ )
327+ session.add(
328+ ConnectorHealth(
329+ service="louka",
330+ source="kangalou",
331+ checked_at=now,
332+ status="broken",
333+ last_success=None,
334+ found_last=0,
335+ median_found=7900.0,
336+ consecutive_failures=3,
337+ message="3 synchros consécutives en échec ou à 0 résultat",
338+ )
339+ )
340+
341+ client = TestClient(app)
342+
343+ body = client.get("/api/v1/monitoring/connectors").json()
344+ assert body["success"] is True
345+ assert body["data"]["summary"]["ok"] == 1
346+ assert body["data"]["summary"]["broken"] == 1
347+ louka = body["data"]["services"]["louka"]
348+ assert louka["app"]["status"] == "ok"
349+ assert louka["connectors"][0]["source"] == "kangalou"
350+
351+ body = client.get("/api/v1/monitoring/connectors/louka").json()
352+ assert body["data"]["service"] == "louka"
353+ assert body["data"]["summary"]["broken"] == 1
354+
355+ assert client.get("/api/v1/monitoring/connectors/nimporte").status_code == 404
356+
357+ # Résumé intégré au /health.
358+ body = client.get("/health").json()
359+ assert body["data"]["connectors"] == {
360+ "ok": 1,
361+ "degraded": 0,
362+ "broken": 1,
363+ "stale": 0,
364+ }
365
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere