mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(gateway): probe the Redis stream bridge and provisioner in /health/ready (#6447)
* fix(gateway): probe the Redis stream bridge and provisioner in /health/ready GET /health/ready only probed the ORM engine and the LangGraph checkpointer. The Redis stream bridge client connects lazily and nothing ever pinged it, so a Gateway whose Redis was down started, answered 200 ready to the Kubernetes readinessProbe and Compose healthcheck, and kept receiving traffic while every run's first publish failed. Since #6328 the Redis bridge is mandatory for multi-instance deployments, so such a replica is not serviceable at all. - StreamBridge gains `async ping() -> bool`; the base returns True (no external backend), RedisStreamBridge round-trips PING and answers False on RedisError. - /health/ready adds a `stream_bridge` probe (not_configured for the memory bridge; unreachable flips the endpoint to 503; a missing bridge fails closed) and a report-only `provisioner` probe of `{sandbox.provisioner_url}/health` that never changes the status code, since every replica shares one provisioner. The provisioner URL is snapshotted on app.state at startup. - A probe that overruns the endpoint deadline is now the only one marked unreachable; finished verdicts are kept. - The public probe gate is one lock per probe kind, so one request's probes still run concurrently while a thundering herd opens at most one probe connection per kind per process. - Tests: bridge ping cases, stream-bridge/provisioner probe cases, verdict and deadline cases, and the first route-level /health/ready tests via TestClient. - Docs: gateway AGENTS.md, CONFIGURATION.md multi-instance notes, Helm README and probe comment, Compose healthcheck comment. * docs(changelog): note the readiness probe fix (#6447) * fix(gateway): apply the sandbox proxy policy to the provisioner probe Review follow-up for #6447. - The provisioner readiness probe built its httpx client with the default trust_env=True. httpx 0.28 treats CIDR entries in NO_PROXY as host patterns rather than subnet bypasses, so with HTTP_PROXY set and NO_PROXY=10.0.0.0/8 a private provisioner that the requests-based sandbox calls reach directly would have been probed through the proxy and reported unreachable on every check. The probe now reuses `sandbox_http_trust_env`: loopback, private, link-local and cluster-local addresses bypass the proxy, external hosts keep the environment's proxy settings. - Pin the liveness/readiness separation with a route test: `GET /health` stays 200 while the stream bridge and database are unreachable and `/health/ready` is 503, so a Redis blip can never become a restart loop.
This commit is contained in:
Nan Gao
GitHub
parent
f154cfd221
commit
5e3c24fe94
15
CHANGELOG.md
15
CHANGELOG.md
@ -804,6 +804,20 @@ This release closes that milestone with **439 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **gateway:** `GET /health/ready` now reports unready while the Redis stream
|
||||
bridge is unreachable. The bridge's Redis client connects lazily and nothing
|
||||
pinged it, so a gateway whose Redis was down started, answered `200 ready` to
|
||||
the Kubernetes readiness probe and the Compose healthcheck, and kept
|
||||
receiving traffic while every run's first publish failed; with the
|
||||
multi-instance gate making the Redis bridge mandatory, such a replica is not
|
||||
serviceable at all. `StreamBridge` gains `ping()`, the response body gains a
|
||||
`stream_bridge` verdict (`not_configured` for the memory bridge) that flips
|
||||
the endpoint to 503 when unreachable, and a report-only `provisioner` verdict
|
||||
from the sandbox provisioner's own `/health` when `sandbox.provisioner_url`
|
||||
is set, which never changes the status code because every replica shares one
|
||||
provisioner. A probe that overruns the endpoint deadline is now the only one
|
||||
marked unreachable, and the public probe gate is one lock per probe kind.
|
||||
([#6447])
|
||||
- **frontend:** Retrying a message after its attachment upload fails now keeps
|
||||
the context that was attached to it. The composer dropped its quotes,
|
||||
conversation references, staged project files and stored draft as soon as a
|
||||
@ -9130,3 +9144,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6407]: https://github.com/bytedance/deer-flow/pull/6407
|
||||
[#6412]: https://github.com/bytedance/deer-flow/pull/6412
|
||||
[#6426]: https://github.com/bytedance/deer-flow/pull/6426
|
||||
[#6447]: https://github.com/bytedance/deer-flow/pull/6447
|
||||
|
||||
@ -44,12 +44,12 @@ locations and keeps `merge_run_context_overrides` clear. Admission and
|
||||
predicate.
|
||||
|
||||
FastAPI listens on port 8001. `GET /health` is liveness; `GET /health/ready`
|
||||
concurrently probes the ORM engine and effective LangGraph checkpointer/Store
|
||||
under one deadline and a strict connection-opening gate. It reads the startup
|
||||
config snapshot: legacy `checkpointer:` if set, else `database:`. Unreachable or
|
||||
unresolved backends return 503; process-local backends such as `memory` report
|
||||
`not_configured`. `GATEWAY_ENABLE_DOCS=false` disables `/docs`, `/redoc`, and
|
||||
`/openapi.json`.
|
||||
concurrently probes the ORM engine, the checkpointer/Store and the stream
|
||||
bridge backend (`StreamBridge.ping()`) under one deadline and per-kind gates,
|
||||
from startup snapshots. Unreachable or unresolved gating backends return 503;
|
||||
`memory` reports `not_configured`; the provisioner `/health` verdict is
|
||||
report-only. See `health.py`. `GATEWAY_ENABLE_DOCS=false` disables `/docs`,
|
||||
`/redoc`, and `/openapi.json`.
|
||||
|
||||
`build_run_config()` resolves the default LangGraph super-step budget from the
|
||||
hot-reloaded top-level `recursion_limit` setting. A valid request-level value
|
||||
|
||||
@ -19,7 +19,7 @@ from app.gateway.browser_capability import ensure_browser_runtime_available
|
||||
from app.gateway.config import get_gateway_config
|
||||
from app.gateway.csrf_middleware import CORS_EXPOSED_HEADERS, CSRFMiddleware, get_configured_cors_origins
|
||||
from app.gateway.deps import langgraph_runtime
|
||||
from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, readiness_payload
|
||||
from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, READINESS_PROVISIONER_URL_ATTR, readiness_payload
|
||||
from app.gateway.routers import (
|
||||
agents,
|
||||
artifacts,
|
||||
@ -1307,19 +1307,26 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for
|
||||
|
||||
@app.get("/health/ready", tags=["health"])
|
||||
async def readiness_check(request: Request, response: Response) -> dict[str, str]:
|
||||
"""Readiness endpoint: 200 when the persistence backends are reachable.
|
||||
"""Readiness endpoint: 200 when the backends behind agent runs are reachable.
|
||||
|
||||
Probes the ORM engine behind ``database:`` and the effective LangGraph
|
||||
Probes the ORM engine behind ``database:``, the effective LangGraph
|
||||
checkpointer/Store backend (legacy ``checkpointer:`` section, otherwise
|
||||
derived from ``database:``) concurrently beneath one bounded deadline.
|
||||
The checkpointer config comes from the startup snapshot recorded by
|
||||
``langgraph_runtime`` (never hot-reloaded config), so orchestrators can
|
||||
gate on the gateway actually being ready rather than merely alive.
|
||||
Returns 503 with ``status: degraded`` when either probe fails or the
|
||||
startup backend cannot be resolved.
|
||||
derived from ``database:``) and the stream bridge's Redis backend
|
||||
concurrently beneath one bounded deadline, and reports the sandbox
|
||||
provisioner's own ``/health`` without gating on it. Targets come from
|
||||
the startup snapshot recorded by ``langgraph_runtime`` (never
|
||||
hot-reloaded config), so orchestrators can gate on the gateway actually
|
||||
being ready rather than merely alive. Returns 503 with
|
||||
``status: degraded`` when a gating probe fails or a startup target
|
||||
cannot be resolved; the provisioner verdict is informational because
|
||||
every replica shares one provisioner.
|
||||
"""
|
||||
checkpointer_config = getattr(request.app.state, READINESS_CHECKPOINTER_CONFIG_ATTR, None)
|
||||
status_code, payload = await readiness_payload(checkpointer_config)
|
||||
state = request.app.state
|
||||
status_code, payload = await readiness_payload(
|
||||
getattr(state, READINESS_CHECKPOINTER_CONFIG_ATTR, None),
|
||||
stream_bridge=getattr(state, "stream_bridge", None),
|
||||
provisioner_url=getattr(state, READINESS_PROVISIONER_URL_ATTR, None),
|
||||
)
|
||||
response.status_code = status_code
|
||||
return payload
|
||||
|
||||
|
||||
@ -580,14 +580,17 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
|
||||
app.state.checkpointer = await stack.enter_async_context(make_checkpointer(config))
|
||||
app.state.store = await stack.enter_async_context(make_store(config))
|
||||
|
||||
# Record the checkpointer/Store backend selected from this startup
|
||||
# snapshot so GET /health/ready probes what the running process
|
||||
# actually uses. These singletons are restart-required by design and
|
||||
# are never rebuilt on config.yaml hot reload, so the probe must not
|
||||
# re-resolve process-wide configuration per request.
|
||||
from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, resolve_checkpointer_config
|
||||
# Record the checkpointer/Store backend and the provisioner endpoint
|
||||
# selected from this startup snapshot so GET /health/ready probes what
|
||||
# the running process actually uses. These singletons are
|
||||
# restart-required by design and are never rebuilt on config.yaml hot
|
||||
# reload, so the probe must not re-resolve process-wide configuration
|
||||
# per request. The stream bridge needs no snapshot: the probe pings the
|
||||
# singleton stored on app.state.stream_bridge above.
|
||||
from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, READINESS_PROVISIONER_URL_ATTR, resolve_checkpointer_config, resolve_provisioner_url
|
||||
|
||||
setattr(app.state, READINESS_CHECKPOINTER_CONFIG_ATTR, resolve_checkpointer_config(config))
|
||||
setattr(app.state, READINESS_PROVISIONER_URL_ATTR, resolve_provisioner_url(config))
|
||||
|
||||
# Initialize repositories — one get_session_factory() call for all.
|
||||
sf = get_session_factory()
|
||||
|
||||
@ -1,29 +1,40 @@
|
||||
"""Readiness probe helpers for the gateway health endpoints.
|
||||
|
||||
``GET /health`` stays a pure liveness signal: 200 whenever the process is up.
|
||||
``GET /health/ready`` additionally probes the persistence the gateway actually
|
||||
``GET /health/ready`` additionally probes the backends the gateway actually
|
||||
depends on, so orchestrators (Docker healthchecks, Kubernetes probes) treat the
|
||||
gateway as ready only when the databases behind agent runs are reachable. Two
|
||||
backends can be configured independently:
|
||||
gateway as ready only when agent runs can be persisted and streamed. Three
|
||||
probes gate the verdict and can be configured independently:
|
||||
|
||||
* the ORM engine behind ``database:`` (application repositories), and
|
||||
* the ORM engine behind ``database:`` (application repositories),
|
||||
* the effective LangGraph checkpointer/Store backend - the legacy
|
||||
``checkpointer:`` section when present, otherwise derived from ``database:``
|
||||
(memory/sqlite/postgres).
|
||||
(memory/sqlite/postgres), and
|
||||
* the stream bridge's external backend (Redis for ``stream_bridge.type:
|
||||
redis``). Its client connects lazily, so without this ping a gateway whose
|
||||
Redis is down starts, reports ready and then fails every run's publish.
|
||||
|
||||
Both probes run concurrently beneath a single endpoint-wide deadline
|
||||
A fourth probe is report-only: when ``sandbox.provisioner_url`` is set, the
|
||||
provisioner's own unauthenticated ``GET /health`` is checked and its verdict
|
||||
is included in the body, but it never flips the endpoint to 503. Every
|
||||
replica shares one provisioner, so a blip there must not drain the service.
|
||||
|
||||
All probes run concurrently beneath a single endpoint-wide deadline
|
||||
(:data:`_READINESS_DEADLINE_SECONDS`), so normal probe work completes within
|
||||
one probe window rather than the sum of both budgets. Connection teardown is
|
||||
ownership-critical and is drained to completion after cancellation, so a
|
||||
stalled close may outlive the probe/deadline budget. The checkpointer config
|
||||
is resolved once at startup from the same snapshot ``langgraph_runtime`` builds
|
||||
its resources from and is stored on ``app.state``; probing a hot-reloaded
|
||||
one probe window rather than the sum of the budgets. A probe that overruns the
|
||||
deadline fails alone; verdicts already collected are kept. Connection teardown
|
||||
is ownership-critical and is drained to completion after cancellation, so a
|
||||
stalled close may outlive the probe/deadline budget. Probe targets are
|
||||
resolved once at startup from the same snapshot ``langgraph_runtime`` builds
|
||||
its resources from and are stored on ``app.state``; probing a hot-reloaded
|
||||
config instead could check a backend the running process is not using. A
|
||||
``backend=memory`` deployment has nothing to probe and is always considered
|
||||
ready; a startup config that cannot be resolved fails closed as unreachable.
|
||||
Connection-opening probes are serialized behind a strict per-process gate: the
|
||||
route is public through the ``/health`` auth prefix, so unlimited concurrent
|
||||
requests must never translate into unlimited new PostgreSQL connections.
|
||||
process-local backend (``memory``) has nothing to probe and reports
|
||||
``not_configured``; a startup target that cannot be resolved fails closed as
|
||||
unreachable. Connection-opening probes are serialized behind strict
|
||||
per-process gates, one per probe kind: the route is public through the
|
||||
``/health`` auth prefix, so unlimited concurrent requests must never translate
|
||||
into unlimited new PostgreSQL connections, Redis pool growth or a connection
|
||||
flood at the provisioner.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -32,8 +43,10 @@ import asyncio
|
||||
import logging
|
||||
import pathlib
|
||||
import weakref
|
||||
from collections.abc import Awaitable
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import text
|
||||
|
||||
from deerflow.persistence.engine import get_engine
|
||||
@ -42,6 +55,7 @@ from deerflow.utils.file_io import await_drained
|
||||
if TYPE_CHECKING:
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.config.checkpointer_config import CheckpointerConfig
|
||||
from deerflow.runtime.stream_bridge import StreamBridge
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -64,46 +78,64 @@ _READINESS_DEADLINE_SECONDS = 3.0
|
||||
# records the startup-bound checkpointer/Store config the probe targets.
|
||||
READINESS_CHECKPOINTER_CONFIG_ATTR = "checkpointer_config"
|
||||
|
||||
# One gate per running event loop (one per worker process in production; one
|
||||
# per test loop in the suite). ``/health/ready`` is public and unauthenticated,
|
||||
# so a thundering herd of probes - or an attacker - must never be able to open
|
||||
# an unbounded number of new connections: every connection-opening probe below
|
||||
# is serialized through this gate, bounding in-flight probe connections to one
|
||||
# per process. Waiting requests are still shed by the endpoint-wide deadline.
|
||||
_PROBE_GATES: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock] = weakref.WeakKeyDictionary()
|
||||
# ``app.state`` attribute under which :func:`app.gateway.deps.langgraph_runtime`
|
||||
# records the startup-bound provisioner base URL (``None`` when unset). The
|
||||
# stream bridge needs no snapshot: the singleton itself lives on
|
||||
# ``app.state.stream_bridge``.
|
||||
READINESS_PROVISIONER_URL_ATTR = "readiness_provisioner_url"
|
||||
|
||||
# Probes whose ``unreachable`` verdict flips the endpoint to 503. The
|
||||
# provisioner is deliberately absent: it is shared by every replica, so its
|
||||
# verdict is informational and must never drain the whole service.
|
||||
_GATING_PROBES = ("database", "checkpointer", "stream_bridge")
|
||||
|
||||
# One set of gates per running event loop (one per worker process in
|
||||
# production; one per test loop in the suite). ``/health/ready`` is public and
|
||||
# unauthenticated, so a thundering herd of probes - or an attacker - must never
|
||||
# be able to open an unbounded number of new connections: every
|
||||
# connection-opening probe below is serialized through the gate for its kind,
|
||||
# bounding in-flight probe connections to one per kind per process. Gates are
|
||||
# per kind rather than shared so the probes of one request still run
|
||||
# concurrently and the endpoint deadline keeps holding for a healthy response.
|
||||
# Waiting requests are still shed by the endpoint-wide deadline.
|
||||
_PROBE_GATES: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = weakref.WeakKeyDictionary()
|
||||
|
||||
|
||||
def _probe_gate() -> asyncio.Lock:
|
||||
"""Return the serialization gate bound to the currently running loop."""
|
||||
def _probe_gate(kind: str) -> asyncio.Lock:
|
||||
"""Return the serialization gate for *kind* bound to the running loop."""
|
||||
loop = asyncio.get_running_loop()
|
||||
gate = _PROBE_GATES.get(loop)
|
||||
gates = _PROBE_GATES.get(loop)
|
||||
if gates is None:
|
||||
gates = {}
|
||||
_PROBE_GATES[loop] = gates
|
||||
gate = gates.get(kind)
|
||||
if gate is None:
|
||||
gate = asyncio.Lock()
|
||||
_PROBE_GATES[loop] = gate
|
||||
gates[kind] = gate
|
||||
return gate
|
||||
|
||||
|
||||
# Result vocabulary for the database probe.
|
||||
DATABASE_OK = "ok"
|
||||
DATABASE_NOT_CONFIGURED = "not_configured"
|
||||
DATABASE_UNREACHABLE = "unreachable"
|
||||
# Result vocabulary shared by every probe.
|
||||
PROBE_OK = "ok"
|
||||
PROBE_NOT_CONFIGURED = "not_configured"
|
||||
PROBE_UNREACHABLE = "unreachable"
|
||||
|
||||
|
||||
async def check_database_health() -> str:
|
||||
"""Probe the persistence engine; return one of the DATABASE_* values."""
|
||||
"""Probe the persistence engine; return one of the PROBE_* values."""
|
||||
engine = get_engine()
|
||||
if engine is None:
|
||||
# backend=memory, or the engine has not been initialized yet: there is
|
||||
# no database to probe.
|
||||
return DATABASE_NOT_CONFIGURED
|
||||
return PROBE_NOT_CONFIGURED
|
||||
try:
|
||||
async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS):
|
||||
async with engine.connect() as connection:
|
||||
await connection.execute(text("SELECT 1"))
|
||||
except Exception:
|
||||
logger.warning("Readiness database probe failed", exc_info=True)
|
||||
return DATABASE_UNREACHABLE
|
||||
return DATABASE_OK
|
||||
return PROBE_UNREACHABLE
|
||||
return PROBE_OK
|
||||
|
||||
|
||||
def resolve_checkpointer_config(startup_config: AppConfig) -> CheckpointerConfig | None:
|
||||
@ -156,16 +188,16 @@ async def _probe_sqlite_backend(conn_string: str | None) -> str:
|
||||
import aiosqlite
|
||||
except ImportError:
|
||||
logger.error("Readiness probe: aiosqlite is not installed for the sqlite checkpointer backend")
|
||||
return DATABASE_UNREACHABLE
|
||||
return PROBE_UNREACHABLE
|
||||
from deerflow.runtime.store._sqlite_utils import resolve_sqlite_conn_str
|
||||
|
||||
try:
|
||||
conn_str = resolve_sqlite_conn_str(conn_string or "store.db")
|
||||
except ValueError as exc:
|
||||
logger.error("Readiness probe: %s", exc)
|
||||
return DATABASE_UNREACHABLE
|
||||
return PROBE_UNREACHABLE
|
||||
if conn_str == ":memory:":
|
||||
return DATABASE_NOT_CONFIGURED
|
||||
return PROBE_NOT_CONFIGURED
|
||||
try:
|
||||
async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS):
|
||||
connection = await aiosqlite.connect(_sqlite_disk_uri(conn_str), uri=True)
|
||||
@ -175,8 +207,8 @@ async def _probe_sqlite_backend(conn_string: str | None) -> str:
|
||||
await await_drained(connection.close())
|
||||
except Exception:
|
||||
logger.warning("Readiness sqlite checkpointer probe failed", exc_info=True)
|
||||
return DATABASE_UNREACHABLE
|
||||
return DATABASE_OK
|
||||
return PROBE_UNREACHABLE
|
||||
return PROBE_OK
|
||||
|
||||
|
||||
async def _probe_postgres_backend(conn_string: str, schema: str) -> str:
|
||||
@ -185,7 +217,7 @@ async def _probe_postgres_backend(conn_string: str, schema: str) -> str:
|
||||
from psycopg import AsyncConnection
|
||||
except ImportError:
|
||||
logger.error("Readiness probe: psycopg is not installed for the postgres checkpointer backend")
|
||||
return DATABASE_UNREACHABLE
|
||||
return PROBE_UNREACHABLE
|
||||
try:
|
||||
from deerflow.persistence.postgres_schema import dsn_with_search_path, normalize_libpq_dsn
|
||||
|
||||
@ -199,8 +231,8 @@ async def _probe_postgres_backend(conn_string: str, schema: str) -> str:
|
||||
await await_drained(connection.close())
|
||||
except Exception:
|
||||
logger.warning("Readiness postgres checkpointer probe failed", exc_info=True)
|
||||
return DATABASE_UNREACHABLE
|
||||
return DATABASE_OK
|
||||
return PROBE_UNREACHABLE
|
||||
return PROBE_OK
|
||||
|
||||
|
||||
async def _probe_checkpointer_backend(config: CheckpointerConfig) -> str:
|
||||
@ -213,33 +245,132 @@ async def _probe_checkpointer_backend(config: CheckpointerConfig) -> str:
|
||||
"""
|
||||
if config.type == "memory":
|
||||
# In-process backend: there is nothing external to probe.
|
||||
return DATABASE_NOT_CONFIGURED
|
||||
return PROBE_NOT_CONFIGURED
|
||||
if config.type not in ("sqlite", "postgres"):
|
||||
logger.warning("Readiness probe: unknown checkpointer backend %r", config.type)
|
||||
return DATABASE_UNREACHABLE
|
||||
async with _probe_gate():
|
||||
return PROBE_UNREACHABLE
|
||||
async with _probe_gate("checkpointer"):
|
||||
if config.type == "sqlite":
|
||||
return await _probe_sqlite_backend(config.connection_string)
|
||||
if not config.connection_string:
|
||||
return DATABASE_UNREACHABLE
|
||||
return PROBE_UNREACHABLE
|
||||
return await _probe_postgres_backend(config.connection_string, config.postgres_schema)
|
||||
|
||||
|
||||
async def readiness_payload(checkpointer_config: CheckpointerConfig | None = None) -> tuple[int, dict[str, str]]:
|
||||
async def _probe_stream_bridge(bridge: StreamBridge | None) -> str:
|
||||
"""Probe the stream bridge's external backend (Redis for the redis bridge).
|
||||
|
||||
``None`` fails closed: without a bridge no run can publish, stream or be
|
||||
cancelled across instances, so readiness must not be claimed. A
|
||||
process-local bridge (``supports_cross_process`` False) has nothing
|
||||
external to probe. Pings are serialized so unauthenticated probe traffic
|
||||
cannot grow the Redis connection pool; the bridge answers ``False`` for a
|
||||
backend-side failure and anything it raises counts the same way.
|
||||
"""
|
||||
if bridge is None:
|
||||
logger.error("Readiness probe: no stream bridge recorded on app.state; failing closed")
|
||||
return PROBE_UNREACHABLE
|
||||
if not bridge.supports_cross_process:
|
||||
return PROBE_NOT_CONFIGURED
|
||||
async with _probe_gate("stream_bridge"):
|
||||
try:
|
||||
async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS):
|
||||
reachable = await bridge.ping()
|
||||
except Exception:
|
||||
logger.warning("Readiness stream bridge probe failed", exc_info=True)
|
||||
return PROBE_UNREACHABLE
|
||||
if not reachable:
|
||||
logger.warning("Readiness stream bridge probe: backend reported unreachable")
|
||||
return PROBE_UNREACHABLE
|
||||
return PROBE_OK
|
||||
|
||||
|
||||
def resolve_provisioner_url(startup_config: AppConfig) -> str | None:
|
||||
"""Return the provisioner base URL from the startup snapshot, or ``None``.
|
||||
|
||||
``sandbox`` is a startup-only section (the provider singleton is built
|
||||
once), so the probe target is pinned at startup like the checkpointer
|
||||
snapshot. ``provisioner_url`` is an ``extra="allow"`` key on
|
||||
``SandboxConfig`` that only ``AioSandboxProvider`` consumes, not a declared
|
||||
field, hence the ``getattr`` access. Blank values mean "no provisioner".
|
||||
"""
|
||||
sandbox = getattr(startup_config, "sandbox", None)
|
||||
url = getattr(sandbox, "provisioner_url", None) if sandbox is not None else None
|
||||
if not isinstance(url, str):
|
||||
return None
|
||||
normalized = url.strip().rstrip("/")
|
||||
return normalized or None
|
||||
|
||||
|
||||
def _make_provisioner_client(provisioner_url: str) -> httpx.AsyncClient:
|
||||
"""Build the short-lived client for one provisioner probe (replaced in tests).
|
||||
|
||||
Proxy handling mirrors the provisioner-bound sandbox clients: a loopback,
|
||||
private, link-local or cluster-local address (``provisioner``,
|
||||
``*.docker.internal``) is control-plane traffic and bypasses ``HTTP_PROXY``,
|
||||
while an external host keeps the environment's proxy settings. Unlike the
|
||||
requests-based sandbox calls, httpx does not honour CIDR entries in
|
||||
``NO_PROXY``, so without this a private provisioner behind a proxy-only
|
||||
egress would be reported unreachable on every probe.
|
||||
"""
|
||||
# Lazy: importing the aio_sandbox package pulls in the whole provider stack.
|
||||
from deerflow.community.aio_sandbox.backend import sandbox_http_trust_env
|
||||
|
||||
return httpx.AsyncClient(timeout=_PROBE_TIMEOUT_SECONDS, trust_env=sandbox_http_trust_env(provisioner_url))
|
||||
|
||||
|
||||
async def _probe_provisioner(provisioner_url: str | None) -> str:
|
||||
"""Probe the sandbox provisioner's unauthenticated ``GET /health``.
|
||||
|
||||
Report-only: :func:`readiness_payload` never degrades on this verdict,
|
||||
because every gateway replica shares one provisioner and a blip there
|
||||
must not drain the whole service. The probe is still serialized per
|
||||
process so public probe traffic cannot become a connection flood at the
|
||||
provisioner, and the per-probe budget bounds a hung connection.
|
||||
"""
|
||||
if not provisioner_url:
|
||||
return PROBE_NOT_CONFIGURED
|
||||
async with _probe_gate("provisioner"):
|
||||
try:
|
||||
async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS):
|
||||
async with _make_provisioner_client(provisioner_url) as client:
|
||||
response = await client.get(f"{provisioner_url}/health")
|
||||
except Exception:
|
||||
logger.warning("Readiness provisioner probe failed", exc_info=True)
|
||||
return PROBE_UNREACHABLE
|
||||
if response.status_code != 200:
|
||||
logger.warning("Readiness provisioner probe: HTTP %s from %s/health", response.status_code, provisioner_url)
|
||||
return PROBE_UNREACHABLE
|
||||
return PROBE_OK
|
||||
|
||||
|
||||
async def readiness_payload(
|
||||
checkpointer_config: CheckpointerConfig | None = None,
|
||||
*,
|
||||
stream_bridge: StreamBridge | None = None,
|
||||
provisioner_url: str | None = None,
|
||||
) -> tuple[int, dict[str, str]]:
|
||||
"""Return the (status_code, body) pair served by ``GET /health/ready``.
|
||||
|
||||
Probes both persistence halves the gateway depends on: the ORM engine
|
||||
behind ``database:`` (repositories) and the effective LangGraph
|
||||
checkpointer/Store backend (the legacy ``checkpointer:`` section, otherwise
|
||||
derived from ``database:``). The probes run concurrently beneath one
|
||||
Probes the backends the gateway depends on: the ORM engine behind
|
||||
``database:`` (repositories), the effective LangGraph checkpointer/Store
|
||||
backend (the legacy ``checkpointer:`` section, otherwise derived from
|
||||
``database:``) and the stream bridge's external backend, plus the
|
||||
report-only provisioner check. The probes run concurrently beneath one
|
||||
endpoint-wide deadline, so normal probe work is bounded by the slowest
|
||||
single probe rather than their sum. Connection teardown is drained after
|
||||
cancellation to preserve ownership and can therefore extend the in-flight
|
||||
request beyond that deadline if close itself stalls. ``checkpointer_config`` is the startup
|
||||
snapshot recorded by ``langgraph_runtime``; None means no snapshot could be
|
||||
resolved, which fails closed as an unreachable backend rather than
|
||||
reporting ready. Either backend can be configured independently of the
|
||||
other, so an unreachable probe on either degrades the endpoint.
|
||||
single probe rather than their sum; a probe that overruns the deadline is
|
||||
the only one marked unreachable, so a wedged report-only probe cannot
|
||||
flip the endpoint through the deadline path. Connection teardown is
|
||||
drained after cancellation to preserve ownership and can therefore extend
|
||||
the in-flight request beyond that deadline if close itself stalls.
|
||||
|
||||
``checkpointer_config`` and ``provisioner_url`` are the startup snapshots
|
||||
recorded by ``langgraph_runtime`` and ``stream_bridge`` is the singleton
|
||||
it built; a ``None`` checkpointer snapshot or bridge fails closed as
|
||||
unreachable rather than reporting ready, while a ``None`` provisioner URL
|
||||
simply means none is configured. Each gating backend can be configured
|
||||
independently of the others, so an unreachable verdict on any of them
|
||||
degrades the endpoint.
|
||||
"""
|
||||
|
||||
async def _probe_engine() -> str:
|
||||
@ -250,23 +381,38 @@ async def readiness_payload(checkpointer_config: CheckpointerConfig | None = Non
|
||||
# Fail closed: without the startup-bound config we cannot know what
|
||||
# backend agent runs use, so readiness must not be claimed.
|
||||
logger.error("Readiness probe: no startup checkpointer config snapshot recorded; failing closed")
|
||||
return DATABASE_UNREACHABLE
|
||||
return PROBE_UNREACHABLE
|
||||
return await _probe_checkpointer_backend(checkpointer_config)
|
||||
|
||||
probes = {
|
||||
"database": _probe_engine(),
|
||||
"checkpointer": _probe_checkpointer(),
|
||||
"stream_bridge": _probe_stream_bridge(stream_bridge),
|
||||
"provisioner": _probe_provisioner(provisioner_url),
|
||||
}
|
||||
# Every verdict starts out unreachable and is overwritten as its probe
|
||||
# finishes, so a probe cancelled by the deadline fails alone instead of
|
||||
# erasing the answers the others already produced.
|
||||
results = dict.fromkeys(probes, PROBE_UNREACHABLE)
|
||||
pending = set(probes)
|
||||
|
||||
async def _record(name: str, probe: Awaitable[str]) -> None:
|
||||
results[name] = await probe
|
||||
pending.discard(name)
|
||||
|
||||
try:
|
||||
async with asyncio.timeout(_READINESS_DEADLINE_SECONDS):
|
||||
database, checkpointer = await asyncio.gather(_probe_engine(), _probe_checkpointer())
|
||||
await asyncio.gather(*(_record(name, probe) for name, probe in probes.items()))
|
||||
except TimeoutError:
|
||||
logger.error(
|
||||
"Readiness probes exceeded the %.1fs endpoint deadline",
|
||||
"Readiness probes exceeded the %.1fs endpoint deadline; failing %s",
|
||||
_READINESS_DEADLINE_SECONDS,
|
||||
", ".join(sorted(pending)),
|
||||
)
|
||||
database = checkpointer = DATABASE_UNREACHABLE
|
||||
degraded = DATABASE_UNREACHABLE in (database, checkpointer)
|
||||
degraded = any(results[name] == PROBE_UNREACHABLE for name in _GATING_PROBES)
|
||||
payload = {
|
||||
"status": "degraded" if degraded else "ready",
|
||||
"service": "deer-flow-gateway",
|
||||
"database": database,
|
||||
"checkpointer": checkpointer,
|
||||
**results,
|
||||
}
|
||||
return (503 if degraded else 200, payload)
|
||||
|
||||
@ -580,6 +580,7 @@ Notes:
|
||||
- `GATEWAY_WORKERS` / `WEB_CONCURRENCY` only count the uvicorn workers of one process tree. A Kubernetes Deployment with `replicas > 1` runs one worker per Pod, so every Pod reports a single worker and the multi-worker startup gate stays inert — while each Pod's startup orphan reconciliation still writes the other Pods' lease-less runs off as crashed on every rolling update.
|
||||
- Set `multi_instance: true` (or export `DEER_FLOW_MULTI_INSTANCE=1`, which lets deploy tooling such as a Helm chart set it from its replica count) on every instance that shares one database. Startup then enforces the same prerequisites as `GATEWAY_WORKERS > 1`: `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true`, and a Redis stream bridge (`stream_bridge.type: redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL`). It refuses an explicit `sandbox.ownership.type: memory`, process-local browser tools, and `scheduler.enabled: true` without `scheduler.multi_instance: true`.
|
||||
- `DEER_FLOW_MULTI_INSTANCE` treats blank, `0`, `false`, `no` and `off` as "not declared"; any other value declares a multi-instance deployment, so a typo fails closed.
|
||||
- `GET /health/ready` pings the Redis stream bridge on every probe and answers 503 (`stream_bridge: unreachable`) while Redis is down, so the orchestrator drains that instance instead of routing it runs it cannot publish or stream; the memory bridge reports `not_configured`. When `sandbox.provisioner_url` is set, the body also carries the provisioner's `/health` verdict (`provisioner: ok|unreachable`), which never changes the status code because every instance shares one provisioner. That probe follows the sandbox clients' proxy policy: loopback, private, link-local and cluster-local provisioner addresses bypass `HTTP_PROXY`, external hosts keep the environment's proxy settings.
|
||||
- The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count.
|
||||
- Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it.
|
||||
|
||||
|
||||
@ -111,5 +111,17 @@ class StreamBridge(abc.ABC):
|
||||
giving late subscribers a chance to drain remaining events.
|
||||
"""
|
||||
|
||||
async def ping(self) -> bool:
|
||||
"""Return whether the bridge's external backend is reachable right now.
|
||||
|
||||
Used by the gateway readiness probe. The default is for bridges with
|
||||
no external backend: the in-process memory bridge has nothing to
|
||||
probe and is always reachable. A bridge that sets
|
||||
:attr:`supports_cross_process` talks to a shared backend and should
|
||||
override this with a bounded round trip, answering ``False`` (not
|
||||
raising) when the backend is down.
|
||||
"""
|
||||
return True
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Release backend resources. Default is a no-op."""
|
||||
|
||||
@ -190,6 +190,20 @@ class RedisStreamBridge(StreamBridge):
|
||||
"""Return whether Redis still has retained stream data for *run_id*."""
|
||||
return bool(await self._redis.exists(self._stream_key(run_id)))
|
||||
|
||||
async def ping(self) -> bool:
|
||||
"""Round-trip ``PING`` to the Redis backend for the readiness probe.
|
||||
|
||||
The client connects lazily, so until something publishes nothing has
|
||||
verified that Redis is reachable; this is the only call that does.
|
||||
Redis-side failures answer ``False`` so the probe gets a verdict rather
|
||||
than a stack of client internals. Callers bound the wait themselves.
|
||||
"""
|
||||
try:
|
||||
return bool(await self._redis.ping())
|
||||
except RedisError:
|
||||
logger.warning("Redis stream bridge ping failed", exc_info=True)
|
||||
return False
|
||||
|
||||
async def _resolve_start_stream_id(self, key: str, last_event_id: str | None) -> str:
|
||||
if last_event_id is None:
|
||||
return "0-0"
|
||||
|
||||
@ -6,20 +6,29 @@ import sqlite3
|
||||
import sys
|
||||
import time
|
||||
from contextlib import asynccontextmanager
|
||||
from types import SimpleNamespace
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import app.gateway.health as health_module
|
||||
from app.gateway.health import (
|
||||
DATABASE_NOT_CONFIGURED,
|
||||
DATABASE_OK,
|
||||
DATABASE_UNREACHABLE,
|
||||
PROBE_NOT_CONFIGURED,
|
||||
PROBE_OK,
|
||||
PROBE_UNREACHABLE,
|
||||
READINESS_CHECKPOINTER_CONFIG_ATTR,
|
||||
READINESS_PROVISIONER_URL_ATTR,
|
||||
_probe_checkpointer_backend,
|
||||
_probe_provisioner,
|
||||
_probe_stream_bridge,
|
||||
check_database_health,
|
||||
readiness_payload,
|
||||
resolve_checkpointer_config,
|
||||
resolve_provisioner_url,
|
||||
)
|
||||
from deerflow.config.checkpointer_config import CheckpointerConfig
|
||||
from deerflow.runtime import MemoryStreamBridge
|
||||
|
||||
|
||||
class _FakeConnection:
|
||||
@ -47,50 +56,84 @@ def _create_sqlite_file(path: pathlib.Path) -> None:
|
||||
sqlite3.connect(str(path)).close()
|
||||
|
||||
|
||||
class _FakeCrossProcessBridge(MemoryStreamBridge):
|
||||
"""Bridge with an external backend whose ``ping`` answer is scripted."""
|
||||
|
||||
supports_cross_process = True
|
||||
|
||||
def __init__(self, *, result: bool = True, error: Exception | None = None, delay: float = 0.0) -> None:
|
||||
super().__init__()
|
||||
self._result = result
|
||||
self._error = error
|
||||
self._delay = delay
|
||||
self.ping_calls = 0
|
||||
|
||||
async def ping(self) -> bool:
|
||||
self.ping_calls += 1
|
||||
if self._delay:
|
||||
await asyncio.sleep(self._delay)
|
||||
if self._error is not None:
|
||||
raise self._error
|
||||
return self._result
|
||||
|
||||
|
||||
def _provisioner_client_factory(handler):
|
||||
"""Return a ``_make_provisioner_client`` replacement backed by ``httpx.MockTransport``."""
|
||||
|
||||
def _factory(provisioner_url: str) -> httpx.AsyncClient:
|
||||
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||
|
||||
return _factory
|
||||
|
||||
|
||||
def _install_provisioner(monkeypatch, handler) -> None:
|
||||
monkeypatch.setattr(health_module, "_make_provisioner_client", _provisioner_client_factory(handler))
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_check_database_health_without_engine(monkeypatch):
|
||||
"""backend=memory (no engine) must report not_configured, never unreachable."""
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: None)
|
||||
|
||||
assert await check_database_health() == DATABASE_NOT_CONFIGURED
|
||||
assert await check_database_health() == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_check_database_health_reachable(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
assert await check_database_health() == DATABASE_OK
|
||||
assert await check_database_health() == PROBE_OK
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_check_database_health_unreachable(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine(unreachable=True))
|
||||
|
||||
assert await check_database_health() == DATABASE_UNREACHABLE
|
||||
assert await check_database_health() == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_ready_when_database_ok_and_memory_backend(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"))
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge())
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["status"] == "ready"
|
||||
assert payload["database"] == DATABASE_OK
|
||||
assert payload["checkpointer"] == DATABASE_NOT_CONFIGURED
|
||||
assert payload["database"] == PROBE_OK
|
||||
assert payload["checkpointer"] == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_degraded_when_database_unreachable(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine(unreachable=True))
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"))
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge())
|
||||
|
||||
assert status_code == 503
|
||||
assert payload["status"] == "degraded"
|
||||
assert payload["database"] == DATABASE_UNREACHABLE
|
||||
assert payload["checkpointer"] == DATABASE_NOT_CONFIGURED
|
||||
assert payload["database"] == PROBE_UNREACHABLE
|
||||
assert payload["checkpointer"] == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -98,12 +141,12 @@ async def test_readiness_payload_ready_when_nothing_configured(monkeypatch):
|
||||
"""backend=memory end to end must stay ready with not_configured results."""
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: None)
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"))
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge())
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["status"] == "ready"
|
||||
assert payload["database"] == DATABASE_NOT_CONFIGURED
|
||||
assert payload["checkpointer"] == DATABASE_NOT_CONFIGURED
|
||||
assert payload["database"] == PROBE_NOT_CONFIGURED
|
||||
assert payload["checkpointer"] == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -112,12 +155,12 @@ async def test_readiness_payload_degraded_when_checkpointer_unreachable_but_data
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
config = CheckpointerConfig(type="sqlite", connection_string=str(tmp_path / "missing" / "checkpoints.db"))
|
||||
|
||||
status_code, payload = await readiness_payload(config)
|
||||
status_code, payload = await readiness_payload(config, stream_bridge=MemoryStreamBridge())
|
||||
|
||||
assert status_code == 503
|
||||
assert payload["status"] == "degraded"
|
||||
assert payload["database"] == DATABASE_OK
|
||||
assert payload["checkpointer"] == DATABASE_UNREACHABLE
|
||||
assert payload["database"] == PROBE_OK
|
||||
assert payload["checkpointer"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -125,12 +168,12 @@ async def test_readiness_payload_fails_closed_without_startup_snapshot(monkeypat
|
||||
"""No startup config snapshot must degrade readiness, never report ready."""
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
status_code, payload = await readiness_payload(None)
|
||||
status_code, payload = await readiness_payload(None, stream_bridge=MemoryStreamBridge())
|
||||
|
||||
assert status_code == 503
|
||||
assert payload["status"] == "degraded"
|
||||
assert payload["database"] == DATABASE_OK
|
||||
assert payload["checkpointer"] == DATABASE_UNREACHABLE
|
||||
assert payload["database"] == PROBE_OK
|
||||
assert payload["checkpointer"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -139,19 +182,26 @@ async def test_readiness_probes_run_concurrently(monkeypatch):
|
||||
|
||||
async def _slow_ok(*args) -> str:
|
||||
await asyncio.sleep(0.35)
|
||||
return DATABASE_OK
|
||||
return PROBE_OK
|
||||
|
||||
monkeypatch.setattr(health_module, "check_database_health", _slow_ok)
|
||||
monkeypatch.setattr(health_module, "_probe_checkpointer_backend", _slow_ok)
|
||||
monkeypatch.setattr(health_module, "_probe_provisioner", _slow_ok)
|
||||
|
||||
started = time.perf_counter()
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"))
|
||||
status_code, payload = await readiness_payload(
|
||||
CheckpointerConfig(type="memory"),
|
||||
stream_bridge=_FakeCrossProcessBridge(delay=0.35),
|
||||
provisioner_url="http://provisioner:8002",
|
||||
)
|
||||
elapsed = time.perf_counter() - started
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["database"] == DATABASE_OK
|
||||
assert payload["checkpointer"] == DATABASE_OK
|
||||
# Two sequential 0.35s probes would take ~0.7s; concurrent ones finish
|
||||
assert payload["database"] == PROBE_OK
|
||||
assert payload["checkpointer"] == PROBE_OK
|
||||
assert payload["stream_bridge"] == PROBE_OK
|
||||
assert payload["provisioner"] == PROBE_OK
|
||||
# Four sequential 0.35s probes would take ~1.4s; concurrent ones finish
|
||||
# within a single probe window.
|
||||
assert elapsed < 0.6
|
||||
|
||||
@ -162,18 +212,18 @@ async def test_readiness_payload_enforces_endpoint_deadline(monkeypatch):
|
||||
|
||||
async def _hanging(*args) -> str:
|
||||
await asyncio.sleep(30)
|
||||
return DATABASE_OK
|
||||
return PROBE_OK
|
||||
|
||||
monkeypatch.setattr(health_module, "check_database_health", _hanging)
|
||||
monkeypatch.setattr(health_module, "_probe_checkpointer_backend", _hanging)
|
||||
monkeypatch.setattr(health_module, "_READINESS_DEADLINE_SECONDS", 0.05)
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"))
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge())
|
||||
|
||||
assert status_code == 503
|
||||
assert payload["status"] == "degraded"
|
||||
assert payload["database"] == DATABASE_UNREACHABLE
|
||||
assert payload["checkpointer"] == DATABASE_UNREACHABLE
|
||||
assert payload["database"] == PROBE_UNREACHABLE
|
||||
assert payload["checkpointer"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -194,7 +244,7 @@ async def test_concurrent_readiness_requests_do_not_open_concurrent_probe_connec
|
||||
max_active = max(max_active, active)
|
||||
try:
|
||||
await asyncio.sleep(0.05)
|
||||
return DATABASE_OK
|
||||
return PROBE_OK
|
||||
finally:
|
||||
active -= 1
|
||||
|
||||
@ -202,7 +252,7 @@ async def test_concurrent_readiness_requests_do_not_open_concurrent_probe_connec
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: None)
|
||||
config = CheckpointerConfig(type="sqlite", connection_string=":memory:")
|
||||
|
||||
results = await asyncio.gather(*(readiness_payload(config) for _ in range(8)))
|
||||
results = await asyncio.gather(*(readiness_payload(config, stream_bridge=MemoryStreamBridge()) for _ in range(8)))
|
||||
|
||||
assert [status_code for status_code, _ in results] == [200] * 8
|
||||
assert max_active == 1
|
||||
@ -210,7 +260,7 @@ async def test_concurrent_readiness_requests_do_not_open_concurrent_probe_connec
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_checkpointer_memory_reports_not_configured():
|
||||
assert await _probe_checkpointer_backend(CheckpointerConfig(type="memory")) == DATABASE_NOT_CONFIGURED
|
||||
assert await _probe_checkpointer_backend(CheckpointerConfig(type="memory")) == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -220,7 +270,7 @@ async def test_probe_checkpointer_sqlite_reachable(tmp_path):
|
||||
|
||||
result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=str(db_path)))
|
||||
|
||||
assert result == DATABASE_OK
|
||||
assert result == PROBE_OK
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -235,7 +285,7 @@ async def test_probe_checkpointer_sqlite_uri_is_unreachable(conn_string, tmp_pat
|
||||
|
||||
result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=conn_string))
|
||||
|
||||
assert result == DATABASE_UNREACHABLE
|
||||
assert result == PROBE_UNREACHABLE
|
||||
assert [path.name for path in tmp_path.iterdir()] == ["checkpoints.db"]
|
||||
|
||||
|
||||
@ -246,7 +296,7 @@ async def test_probe_checkpointer_sqlite_missing_file_stays_missing_and_unreacha
|
||||
|
||||
result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=str(missing)))
|
||||
|
||||
assert result == DATABASE_UNREACHABLE
|
||||
assert result == PROBE_UNREACHABLE
|
||||
assert not missing.exists()
|
||||
|
||||
|
||||
@ -256,7 +306,7 @@ async def test_probe_checkpointer_sqlite_unreachable_when_parent_missing(tmp_pat
|
||||
|
||||
result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=str(missing_parent)))
|
||||
|
||||
assert result == DATABASE_UNREACHABLE
|
||||
assert result == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -264,7 +314,7 @@ async def test_probe_checkpointer_sqlite_in_memory_is_not_configured():
|
||||
"""In-memory SQLite has no external state, mirroring the memory backend."""
|
||||
result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=":memory:"))
|
||||
|
||||
assert result == DATABASE_NOT_CONFIGURED
|
||||
assert result == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -278,7 +328,7 @@ async def test_probe_checkpointer_postgres_without_psycopg_is_unreachable(monkey
|
||||
)
|
||||
)
|
||||
|
||||
assert result == DATABASE_UNREACHABLE
|
||||
assert result == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
def test_resolve_checkpointer_config_passes_through_resolution(monkeypatch):
|
||||
@ -399,3 +449,389 @@ async def test_postgres_probe_drains_connection_close_across_repeated_cancellati
|
||||
connection,
|
||||
"PostgreSQL",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stream bridge probe
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_stream_bridge_memory_reports_not_configured():
|
||||
"""The process-local memory bridge has no external backend to probe."""
|
||||
bridge = MemoryStreamBridge()
|
||||
|
||||
assert await _probe_stream_bridge(bridge) == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_stream_bridge_reachable_backend_is_ok():
|
||||
bridge = _FakeCrossProcessBridge(result=True)
|
||||
|
||||
assert await _probe_stream_bridge(bridge) == PROBE_OK
|
||||
assert bridge.ping_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_stream_bridge_false_ping_is_unreachable():
|
||||
assert await _probe_stream_bridge(_FakeCrossProcessBridge(result=False)) == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_stream_bridge_raising_ping_is_unreachable():
|
||||
assert await _probe_stream_bridge(_FakeCrossProcessBridge(error=RuntimeError("redis exploded"))) == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_stream_bridge_enforces_probe_timeout(monkeypatch):
|
||||
"""A ping that never answers must be bounded by the per-probe budget."""
|
||||
monkeypatch.setattr(health_module, "_PROBE_TIMEOUT_SECONDS", 0.05)
|
||||
|
||||
started = time.perf_counter()
|
||||
result = await _probe_stream_bridge(_FakeCrossProcessBridge(delay=30))
|
||||
elapsed = time.perf_counter() - started
|
||||
|
||||
assert result == PROBE_UNREACHABLE
|
||||
assert elapsed < 1.0
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_stream_bridge_fails_closed_without_bridge():
|
||||
"""No bridge on app.state means runs cannot stream: never report ready."""
|
||||
assert await _probe_stream_bridge(None) == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_concurrent_stream_bridge_probes_are_serialized():
|
||||
"""Public /health/ready must not fan a thundering herd out onto the Redis pool."""
|
||||
active = 0
|
||||
max_active = 0
|
||||
|
||||
class _TrackingBridge(_FakeCrossProcessBridge):
|
||||
async def ping(self) -> bool:
|
||||
nonlocal active, max_active
|
||||
active += 1
|
||||
max_active = max(max_active, active)
|
||||
try:
|
||||
await asyncio.sleep(0.02)
|
||||
return True
|
||||
finally:
|
||||
active -= 1
|
||||
|
||||
bridge = _TrackingBridge()
|
||||
results = await asyncio.gather(*(_probe_stream_bridge(bridge) for _ in range(6)))
|
||||
|
||||
assert results == [PROBE_OK] * 6
|
||||
assert max_active == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Provisioner probe
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_resolve_provisioner_url_normalises_configured_url():
|
||||
config = SimpleNamespace(sandbox=SimpleNamespace(provisioner_url=" http://provisioner:8002/ "))
|
||||
|
||||
assert resolve_provisioner_url(config) == "http://provisioner:8002"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config",
|
||||
[
|
||||
SimpleNamespace(sandbox=SimpleNamespace()),
|
||||
SimpleNamespace(sandbox=SimpleNamespace(provisioner_url=None)),
|
||||
SimpleNamespace(sandbox=SimpleNamespace(provisioner_url=" ")),
|
||||
SimpleNamespace(sandbox=None),
|
||||
SimpleNamespace(),
|
||||
],
|
||||
ids=["no-field", "none", "blank", "no-sandbox", "no-sandbox-attr"],
|
||||
)
|
||||
def test_resolve_provisioner_url_absent(config):
|
||||
assert resolve_provisioner_url(config) is None
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_provisioner_not_configured_without_url():
|
||||
assert await _probe_provisioner(None) == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize(
|
||||
("provisioner_url", "trust_env"),
|
||||
[
|
||||
("http://provisioner:8002", False),
|
||||
("http://10.1.2.3:8002", False),
|
||||
("http://127.0.0.1:8002", False),
|
||||
("http://provisioner.docker.internal:8002", False),
|
||||
("https://provisioner.example.com", True),
|
||||
("http://8.8.8.8:8002", True),
|
||||
],
|
||||
ids=["service-name", "private-ip", "loopback", "docker-internal", "external-fqdn", "public-ip"],
|
||||
)
|
||||
async def test_make_provisioner_client_bypasses_proxy_for_control_plane_addresses(provisioner_url, trust_env):
|
||||
"""Control-plane targets must not be routed through HTTP_PROXY.
|
||||
|
||||
httpx does not treat CIDR entries in NO_PROXY as subnet bypasses the way the
|
||||
requests-based sandbox calls do, so the probe applies the same
|
||||
target-aware policy those clients use instead of trusting the environment.
|
||||
"""
|
||||
async with health_module._make_provisioner_client(provisioner_url) as client:
|
||||
assert client.trust_env is trust_env
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_provisioner_calls_unauthenticated_health_route(monkeypatch):
|
||||
"""The probe targets the provisioner's own GET /health, which needs no API key."""
|
||||
seen: list[httpx.Request] = []
|
||||
|
||||
def _handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request)
|
||||
return httpx.Response(200, json={"status": "ok"})
|
||||
|
||||
_install_provisioner(monkeypatch, _handler)
|
||||
|
||||
assert await _probe_provisioner("http://provisioner:8002") == PROBE_OK
|
||||
assert [str(request.url) for request in seen] == ["http://provisioner:8002/health"]
|
||||
assert "x-api-key" not in {name.lower() for name in seen[0].headers}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_provisioner_non_200_is_unreachable(monkeypatch):
|
||||
_install_provisioner(monkeypatch, lambda request: httpx.Response(503, text="draining"))
|
||||
|
||||
assert await _probe_provisioner("http://provisioner:8002") == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_provisioner_connection_error_is_unreachable(monkeypatch):
|
||||
def _handler(request: httpx.Request) -> httpx.Response:
|
||||
raise httpx.ConnectError("connection refused", request=request)
|
||||
|
||||
_install_provisioner(monkeypatch, _handler)
|
||||
|
||||
assert await _probe_provisioner("http://provisioner:8002") == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_probe_provisioner_enforces_probe_timeout(monkeypatch):
|
||||
monkeypatch.setattr(health_module, "_PROBE_TIMEOUT_SECONDS", 0.05)
|
||||
|
||||
async def _hanging_handler(request: httpx.Request) -> httpx.Response:
|
||||
await asyncio.sleep(30)
|
||||
return httpx.Response(200)
|
||||
|
||||
_install_provisioner(monkeypatch, _hanging_handler)
|
||||
|
||||
started = time.perf_counter()
|
||||
result = await _probe_provisioner("http://provisioner:8002")
|
||||
elapsed = time.perf_counter() - started
|
||||
|
||||
assert result == PROBE_UNREACHABLE
|
||||
assert elapsed < 1.0
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_concurrent_provisioner_probes_are_serialized(monkeypatch):
|
||||
"""Public /health/ready must not amplify into a connection flood at the provisioner."""
|
||||
active = 0
|
||||
max_active = 0
|
||||
|
||||
async def _handler(request: httpx.Request) -> httpx.Response:
|
||||
nonlocal active, max_active
|
||||
active += 1
|
||||
max_active = max(max_active, active)
|
||||
try:
|
||||
await asyncio.sleep(0.02)
|
||||
return httpx.Response(200)
|
||||
finally:
|
||||
active -= 1
|
||||
|
||||
_install_provisioner(monkeypatch, _handler)
|
||||
|
||||
results = await asyncio.gather(*(_probe_provisioner("http://provisioner:8002") for _ in range(6)))
|
||||
|
||||
assert results == [PROBE_OK] * 6
|
||||
assert max_active == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# readiness_payload: stream bridge and provisioner verdicts
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_degraded_when_stream_bridge_unreachable(monkeypatch):
|
||||
"""Redis down means no run can publish, stream or be cancelled: the pod is not ready."""
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=_FakeCrossProcessBridge(result=False))
|
||||
|
||||
assert status_code == 503
|
||||
assert payload["status"] == "degraded"
|
||||
assert payload["database"] == PROBE_OK
|
||||
assert payload["checkpointer"] == PROBE_NOT_CONFIGURED
|
||||
assert payload["stream_bridge"] == PROBE_UNREACHABLE
|
||||
assert payload["provisioner"] == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_ready_when_stream_bridge_reachable(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=_FakeCrossProcessBridge(result=True))
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["status"] == "ready"
|
||||
assert payload["stream_bridge"] == PROBE_OK
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_fails_closed_without_stream_bridge(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=None)
|
||||
|
||||
assert status_code == 503
|
||||
assert payload["stream_bridge"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_reports_unreachable_provisioner_without_degrading(monkeypatch):
|
||||
"""Every replica shares one provisioner; its blip must not take the whole service out."""
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
_install_provisioner(monkeypatch, lambda request: httpx.Response(503))
|
||||
|
||||
status_code, payload = await readiness_payload(
|
||||
CheckpointerConfig(type="memory"),
|
||||
stream_bridge=MemoryStreamBridge(),
|
||||
provisioner_url="http://provisioner:8002",
|
||||
)
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["status"] == "ready"
|
||||
assert payload["provisioner"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_payload_reports_reachable_provisioner(monkeypatch):
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
_install_provisioner(monkeypatch, lambda request: httpx.Response(200, json={"status": "ok"}))
|
||||
|
||||
status_code, payload = await readiness_payload(
|
||||
CheckpointerConfig(type="memory"),
|
||||
stream_bridge=MemoryStreamBridge(),
|
||||
provisioner_url="http://provisioner:8002",
|
||||
)
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["provisioner"] == PROBE_OK
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_readiness_deadline_only_fails_the_probes_that_overran(monkeypatch):
|
||||
"""A probe that ignores its budget must not erase the verdicts of the probes that finished.
|
||||
|
||||
Otherwise a wedged report-only provisioner probe would flip the whole
|
||||
endpoint to 503 through the deadline path.
|
||||
"""
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
|
||||
async def _hanging(*args) -> str:
|
||||
await asyncio.sleep(30)
|
||||
return PROBE_OK
|
||||
|
||||
monkeypatch.setattr(health_module, "_probe_provisioner", _hanging)
|
||||
monkeypatch.setattr(health_module, "_READINESS_DEADLINE_SECONDS", 0.1)
|
||||
|
||||
status_code, payload = await readiness_payload(
|
||||
CheckpointerConfig(type="memory"),
|
||||
stream_bridge=MemoryStreamBridge(),
|
||||
provisioner_url="http://provisioner:8002",
|
||||
)
|
||||
|
||||
assert status_code == 200
|
||||
assert payload["status"] == "ready"
|
||||
assert payload["database"] == PROBE_OK
|
||||
assert payload["checkpointer"] == PROBE_NOT_CONFIGURED
|
||||
assert payload["stream_bridge"] == PROBE_NOT_CONFIGURED
|
||||
assert payload["provisioner"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTTP route
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _ready_client(monkeypatch, *, stream_bridge, provisioner_url: str | None) -> TestClient:
|
||||
"""Build the real gateway app and wire the startup state the probe reads.
|
||||
|
||||
The lifespan is deliberately not run: the test pins the ``app.state``
|
||||
contract the route depends on instead of booting persistence.
|
||||
"""
|
||||
from app.gateway.app import create_app
|
||||
|
||||
app = create_app()
|
||||
setattr(app.state, READINESS_CHECKPOINTER_CONFIG_ATTR, CheckpointerConfig(type="memory"))
|
||||
setattr(app.state, READINESS_PROVISIONER_URL_ATTR, provisioner_url)
|
||||
app.state.stream_bridge = stream_bridge
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine())
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_health_ready_route_reports_every_dependency(monkeypatch):
|
||||
_install_provisioner(monkeypatch, lambda request: httpx.Response(200, json={"status": "ok"}))
|
||||
client = _ready_client(monkeypatch, stream_bridge=_FakeCrossProcessBridge(result=True), provisioner_url="http://provisioner:8002")
|
||||
|
||||
response = client.get("/health/ready")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {
|
||||
"status": "ready",
|
||||
"service": "deer-flow-gateway",
|
||||
"database": PROBE_OK,
|
||||
"checkpointer": PROBE_NOT_CONFIGURED,
|
||||
"stream_bridge": PROBE_OK,
|
||||
"provisioner": PROBE_OK,
|
||||
}
|
||||
|
||||
|
||||
def test_health_ready_route_returns_503_when_stream_bridge_unreachable(monkeypatch):
|
||||
client = _ready_client(monkeypatch, stream_bridge=_FakeCrossProcessBridge(result=False), provisioner_url=None)
|
||||
|
||||
response = client.get("/health/ready")
|
||||
|
||||
assert response.status_code == 503
|
||||
body = response.json()
|
||||
assert body["status"] == "degraded"
|
||||
assert body["stream_bridge"] == PROBE_UNREACHABLE
|
||||
assert body["provisioner"] == PROBE_NOT_CONFIGURED
|
||||
|
||||
|
||||
def test_health_ready_route_keeps_200_when_only_provisioner_is_down(monkeypatch):
|
||||
def _handler(request: httpx.Request) -> httpx.Response:
|
||||
raise httpx.ConnectError("connection refused", request=request)
|
||||
|
||||
_install_provisioner(monkeypatch, _handler)
|
||||
client = _ready_client(monkeypatch, stream_bridge=MemoryStreamBridge(), provisioner_url="http://provisioner:8002")
|
||||
|
||||
response = client.get("/health/ready")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["provisioner"] == PROBE_UNREACHABLE
|
||||
|
||||
|
||||
def test_health_liveness_route_stays_200_while_readiness_is_degraded(monkeypatch):
|
||||
"""``GET /health`` is liveness only: an unreachable bridge or database must never restart the pod.
|
||||
|
||||
Readiness pulling the pod out of the Service is the intended reaction to a
|
||||
Redis or database outage; the liveness probe moving with it would turn the
|
||||
same outage into a restart loop.
|
||||
"""
|
||||
client = _ready_client(monkeypatch, stream_bridge=_FakeCrossProcessBridge(result=False), provisioner_url=None)
|
||||
monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine(unreachable=True))
|
||||
|
||||
assert client.get("/health/ready").status_code == 503
|
||||
live = client.get("/health")
|
||||
assert live.status_code == 200
|
||||
assert live.json() == {"status": "healthy", "service": "deer-flow-gateway"}
|
||||
|
||||
@ -38,6 +38,14 @@ class _FakeRedis:
|
||||
self.deleted = []
|
||||
self.expirations = []
|
||||
self.closed = False
|
||||
self.ping_calls = 0
|
||||
self.ping_error: Exception | None = None
|
||||
|
||||
async def ping(self):
|
||||
self.ping_calls += 1
|
||||
if self.ping_error is not None:
|
||||
raise self.ping_error
|
||||
return True
|
||||
|
||||
async def xadd(self, name, fields, maxlen=None, approximate=True):
|
||||
self.counters[name] += 1
|
||||
@ -535,6 +543,42 @@ def redis_bridge() -> RedisStreamBridge:
|
||||
return RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=_FakeRedis())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Readiness ping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_memory_bridge_ping_has_no_external_backend():
|
||||
"""The in-process bridge has nothing external to probe and never reports unreachable."""
|
||||
bridge = MemoryStreamBridge()
|
||||
|
||||
assert bridge.supports_cross_process is False
|
||||
assert await bridge.ping() is True
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_redis_bridge_ping_reports_reachable_backend():
|
||||
fake = _FakeRedis()
|
||||
bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=fake)
|
||||
|
||||
assert bridge.supports_cross_process is True
|
||||
assert await bridge.ping() is True
|
||||
assert fake.ping_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_redis_bridge_ping_reports_unreachable_backend_on_redis_error():
|
||||
"""A Redis-side failure is a readiness answer (False), not an exception for the probe to unpack."""
|
||||
from redis.exceptions import ConnectionError as RedisConnectionError
|
||||
|
||||
fake = _FakeRedis()
|
||||
fake.ping_error = RedisConnectionError("connection refused")
|
||||
bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=fake)
|
||||
|
||||
assert await bridge.ping() is False
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_redis_publish_subscribe(redis_bridge: RedisStreamBridge):
|
||||
"""Redis bridge should deliver events in order and terminate on end."""
|
||||
|
||||
@ -234,9 +234,16 @@ helm install deer-flow deploy/helm/deer-flow \
|
||||
```bash
|
||||
kubectl -n deer-flow get pods
|
||||
kubectl -n deer-flow port-forward svc/nginx 2026:2026
|
||||
curl http://localhost:2026/health # gateway health via nginx
|
||||
curl http://localhost:2026/health # gateway liveness via nginx
|
||||
curl http://localhost:2026/health/ready # readiness: database, checkpointer, stream_bridge, provisioner
|
||||
```
|
||||
|
||||
`/health/ready` is what the gateway `readinessProbe` hits. It answers 503 while
|
||||
Postgres or the Redis stream bridge is unreachable (`stream_bridge: unreachable`),
|
||||
so those pods leave the Service instead of accepting runs they cannot stream.
|
||||
The `provisioner` field reports the provisioner's own `/health` but never
|
||||
changes the status code: every gateway pod shares that one provisioner.
|
||||
|
||||
Hit the Ingress host (map it in `/etc/hosts` for local clusters) to load the UI.
|
||||
|
||||
Provisioner sanity check (`/health` is unauthenticated; `/api/*` requires the
|
||||
@ -354,7 +361,10 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002
|
||||
`DEER_FLOW_STREAM_BRIDGE_REDIS_URL`; `config.yaml` sets `stream_bridge.type:
|
||||
redis` by default. No-auth by default (ClusterIP isolation, matching compose);
|
||||
set `redis.auth.password` to enable AUTH. For a managed Redis, disable the
|
||||
bundled instance and point at it via `redis.external`.
|
||||
bundled instance and point at it via `redis.external`. The gateway readiness
|
||||
probe pings this Redis on every check and reports the pod unready while it
|
||||
is unreachable, since without the bridge no run can publish, stream or be
|
||||
cancelled from a peer pod.
|
||||
- **Persistence.** A PVC (`<release>-home`) backs `/app/backend/.deer-flow`
|
||||
(sqlite DB, memory, custom agents, per-thread user-data). The gateway mounts
|
||||
it with `subPath: deer-flow` so the layout matches the provisioner's PVC
|
||||
|
||||
@ -216,9 +216,12 @@ spec:
|
||||
port: http
|
||||
initialDelaySeconds: 5
|
||||
periodSeconds: 10
|
||||
# /health/ready bounds both probes to a 3s endpoint deadline;
|
||||
# /health/ready bounds all its probes (database, checkpointer,
|
||||
# Redis stream bridge, provisioner) to a 3s endpoint deadline;
|
||||
# timeoutSeconds must stay above that bound or Kubernetes aborts
|
||||
# slow-but-healthy responses at its 1s default.
|
||||
# slow-but-healthy responses at its 1s default. An unreachable
|
||||
# Redis makes the pod unready; the provisioner verdict is
|
||||
# report-only.
|
||||
timeoutSeconds: 5
|
||||
livenessProbe:
|
||||
httpGet:
|
||||
|
||||
@ -180,8 +180,10 @@ services:
|
||||
redis:
|
||||
condition: service_healthy
|
||||
healthcheck:
|
||||
# /health/ready runs both probes concurrently within a 3s endpoint
|
||||
# deadline; the client timeout must stay above that bound.
|
||||
# /health/ready runs its probes (database, checkpointer, Redis stream
|
||||
# bridge, provisioner) concurrently within a 3s endpoint deadline; the
|
||||
# client timeout must stay above that bound. The gateway is unhealthy
|
||||
# while Redis is unreachable; the provisioner verdict is report-only.
|
||||
test: ["CMD", "python", "-c", "import urllib.request; response = urllib.request.urlopen('http://127.0.0.1:8001/health/ready', timeout=5); raise SystemExit(0 if response.status == 200 else 1)"]
|
||||
interval: 5s
|
||||
timeout: 5s
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user