This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new b77ce8e feat: separate liveness and readiness probes (#131)
b77ce8e is described below
commit b77ce8e69a0a7cc53576733d1b6d8cdbfbd73239
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 01:46:10 2026 +0800
feat: separate liveness and readiness probes (#131)
---
README.md | 22 ++-
doris_mcp_server/auth/auth_middleware.py | 13 +-
doris_mcp_server/auth/operation_policy.py | 2 +
doris_mcp_server/health.py | 90 +++++++++
doris_mcp_server/main.py | 36 +++-
doris_mcp_server/multiworker_app.py | 70 ++++++-
doris_mcp_server/utils/db.py | 27 ++-
test/integration/test_real_doris_transports.py | 49 ++++-
test/protocol/test_health_endpoints.py | 250 +++++++++++++++++++++++++
test/protocol/test_multiworker_config.py | 47 ++++-
10 files changed, 570 insertions(+), 36 deletions(-)
diff --git a/README.md b/README.md
index ac44bbc..edc2123 100644
--- a/README.md
+++ b/README.md
@@ -189,7 +189,8 @@ doris-mcp-server --help
doris-mcp-client --help
# Test HTTP mode (in another terminal)
-curl http://localhost:3000/health
+curl --fail http://localhost:3000/live
+curl --fail http://localhost:3000/ready
```
### Environment Variables (Optional)
@@ -383,7 +384,14 @@ docker run -d -p <port>:<port> -v
/*your-host*/doris-mcp-server/.env:/app/.env -
**Service Endpoints:**
* **Streamable HTTP**: `http://<host>:<port>/mcp` (MCP messages use `POST`;
do not depend on `GET` or `DELETE` compatibility behavior)
-* **Health Check**: `http://<host>:<port>/health`
+* **Liveness**: `http://<host>:<port>/live` — process and protocol service
are running; does not depend on Doris
+* **Readiness**: `http://<host>:<port>/ready` — returns 200 only when a
bounded Doris `SELECT 1` probe succeeds; otherwise returns 503
+* **Legacy Health Check**: `http://<host>:<port>/health` —
backward-compatible liveness alias; do not use it to decide whether to route
database work
+
+The readiness probe has a fixed short timeout and exposes only stable status
+fields, not connection errors or credentials. Stdio mode has no HTTP health
+surface; supervise the process and use the MCP initialization/discovery
+handshake for transport-level availability.
> **Note**: The server uses Streamable HTTP for web-based communication,
> providing unified request/response and streaming capabilities.
@@ -1568,8 +1576,9 @@ Recommendations:
2. **Check network connectivity**:
```bash
- # Test database connection
- curl http://localhost:3000/health
+ # /live verifies the process; /ready also verifies Doris
+ curl --fail http://localhost:3000/live
+ curl --fail http://localhost:3000/ready
```
3. **Optimize connection pool configuration**:
@@ -1615,8 +1624,9 @@ tail -f logs/doris_mcp_server_info.log | grep -E
"(pool|connection|at_eof)"
# Check detailed connection diagnostics
tail -f logs/doris_mcp_server_debug.log | grep "connection health"
-# View connection pool metrics
-curl http://localhost:8000/health # If running in HTTP mode
+# Check process liveness and Doris readiness
+curl --fail http://localhost:8000/live
+curl --fail http://localhost:8000/ready
```
#### Configuration for Optimal Connection Performance:
diff --git a/doris_mcp_server/auth/auth_middleware.py
b/doris_mcp_server/auth/auth_middleware.py
index 4e7137e..ab6dd91 100644
--- a/doris_mcp_server/auth/auth_middleware.py
+++ b/doris_mcp_server/auth/auth_middleware.py
@@ -152,7 +152,13 @@ class AuthMiddleware:
Returns:
ASGI middleware function
"""
- skip_paths = skip_paths or ['/health', '/docs', '/openapi.json']
+ skip_paths = skip_paths or [
+ '/health',
+ '/live',
+ '/ready',
+ '/docs',
+ '/openapi.json',
+ ]
async def middleware(scope, receive, send):
"""HTTP authentication middleware"""
@@ -163,7 +169,10 @@ class AuthMiddleware:
path = scope.get('path', '')
# Check if authentication should be skipped
- if any(path.startswith(skip) for skip in skip_paths):
+ if any(
+ path == skip or path.startswith(f"{skip}/")
+ for skip in skip_paths
+ ):
return await self.app(scope, receive, send)
# Extract authentication information
diff --git a/doris_mcp_server/auth/operation_policy.py
b/doris_mcp_server/auth/operation_policy.py
index d97f10b..ee52b68 100644
--- a/doris_mcp_server/auth/operation_policy.py
+++ b/doris_mcp_server/auth/operation_policy.py
@@ -338,6 +338,8 @@ OPERATION_POLICIES: dict[str, OperationPolicy] = {
"get_prompt": OperationPolicy("get_prompt", "prompt:get", "deny", "mysql",
"medium"),
"http:/": OperationPolicy("http:/", None, "allow"),
"http:/health": OperationPolicy("http:/health", None, "allow"),
+ "http:/live": OperationPolicy("http:/live", None, "allow"),
+ "http:/ready": OperationPolicy("http:/ready", None, "allow"),
"http:/auth/login": OperationPolicy("http:/auth/login", None, "deny",
"external_oauth", "medium"),
"http:/auth/callback": OperationPolicy("http:/auth/callback", None,
"deny", "external_oauth", "medium"),
"http:/auth/provider": OperationPolicy("http:/auth/provider", None,
"deny", "external_oauth", "medium"),
diff --git a/doris_mcp_server/health.py b/doris_mcp_server/health.py
new file mode 100644
index 0000000..8503d4f
--- /dev/null
+++ b/doris_mcp_server/health.py
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Liveness and readiness response helpers."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Mapping
+from typing import Any, Protocol
+
+DEFAULT_READINESS_TIMEOUT_SECONDS = 2.0
+
+
+class ReadinessProbe(Protocol):
+ """Minimal dependency contract used by the HTTP readiness endpoint."""
+
+ async def check_readiness(self, *, timeout_seconds: float) -> bool: ...
+
+
+def liveness_payload(
+ *,
+ service: str,
+ version: str,
+ legacy: bool = False,
+ details: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+ """Build a database-independent liveness payload."""
+ payload: dict[str, Any] = {
+ "status": "healthy" if legacy else "alive",
+ "service": service,
+ "version": version,
+ "checks": {"process": "alive"},
+ }
+ if details:
+ payload.update(details)
+ return payload
+
+
+async def readiness_payload(
+ connection_manager: ReadinessProbe | None,
+ *,
+ service: str,
+ version: str,
+ initialized: bool = True,
+ details: Mapping[str, Any] | None = None,
+ timeout_seconds: float = DEFAULT_READINESS_TIMEOUT_SECONDS,
+) -> tuple[dict[str, Any], int]:
+ """Build a bounded readiness payload and its HTTP status code."""
+ doris_ready = False
+ if initialized and connection_manager is not None:
+ try:
+ doris_ready = bool(
+ await asyncio.wait_for(
+ connection_manager.check_readiness(
+ timeout_seconds=timeout_seconds,
+ ),
+ timeout=timeout_seconds,
+ )
+ )
+ except Exception:
+ doris_ready = False
+
+ ready = initialized and doris_ready
+ payload: dict[str, Any] = {
+ "status": "ready" if ready else "not_ready",
+ "service": service,
+ "version": version,
+ "checks": {
+ "service": "ready" if initialized else "not_ready",
+ "doris": "ready" if doris_ready else "not_ready",
+ },
+ "timeout_seconds": timeout_seconds,
+ }
+ if details:
+ payload.update(details)
+ return payload, 200 if ready else 503
diff --git a/doris_mcp_server/main.py b/doris_mcp_server/main.py
index f53aca1..7102194 100644
--- a/doris_mcp_server/main.py
+++ b/doris_mcp_server/main.py
@@ -29,6 +29,7 @@ import os
import sys
from ._version import __version__
+from .health import liveness_payload, readiness_payload
from .protocol import create_doris_mcp_server, create_transport_security
from .tools.prompts_manager import DorisPromptsManager
from .tools.resources_manager import DorisResourcesManager
@@ -225,12 +226,28 @@ class DorisServer:
# Health check endpoint
async def health_check(request):
return JSONResponse(
- {
- "status": "healthy",
- "service": self.config.server_name,
- "version": __version__,
- }
+ liveness_payload(
+ service=self.config.server_name,
+ version=__version__,
+ legacy=True,
+ )
+ )
+
+ async def live_check(request):
+ return JSONResponse(
+ liveness_payload(
+ service=self.config.server_name,
+ version=__version__,
+ )
+ )
+
+ async def readiness_check(request):
+ payload, status_code = await readiness_payload(
+ self.connection_manager,
+ service=self.config.server_name,
+ version=__version__,
)
+ return JSONResponse(payload, status_code=status_code)
# OAuth endpoints
from .auth.oauth_handlers import OAuthHandlers
@@ -293,7 +310,11 @@ class DorisServer:
self.logger.info("Application is shutting down...")
effective_auth = get_effective_auth_config(self.config)
- routes = [Route("/health", health_check, methods=["GET"])]
+ routes = [
+ Route("/health", health_check, methods=["GET"]),
+ Route("/live", live_check, methods=["GET"]),
+ Route("/ready", readiness_check, methods=["GET"]),
+ ]
if effective_auth.enable_external_oauth_auth:
routes.extend([
Route(
@@ -355,7 +376,8 @@ class DorisServer:
await response(scope, receive, send)
return
- if (path.startswith("/health") or
+ if (
+ path.rstrip("/") in {"/health", "/live", "/ready"}
or
path.startswith("/auth/") or
path.startswith("/token/") or
path.startswith("/.well-known/") or
diff --git a/doris_mcp_server/multiworker_app.py
b/doris_mcp_server/multiworker_app.py
index 89b924c..74d1335 100644
--- a/doris_mcp_server/multiworker_app.py
+++ b/doris_mcp_server/multiworker_app.py
@@ -31,6 +31,7 @@ from starlette.responses import JSONResponse
from starlette.routing import Route
from ._version import __version__
+from .health import liveness_payload, readiness_payload
from .protocol import create_doris_mcp_server, create_transport_security
from .tools.prompts_manager import DorisPromptsManager
from .tools.resources_manager import DorisResourcesManager
@@ -95,7 +96,17 @@ async def initialize_worker():
_worker_security_manager.connection_manager =
_worker_connection_manager
_worker_security_manager.auth_provider.configure_doris_oauth(_worker_connection_manager)
- await _worker_connection_manager.initialize()
+ global_pool_created = (
+ await _worker_connection_manager.initialize_for_http_mode()
+ )
+ if (
+ not global_pool_created
+ and _worker_effective_auth.enable_doris_oauth_auth
+ ):
+ raise RuntimeError(
+ "Doris OAuth requires the configured service/global Doris
account "
+ "to initialize successfully"
+ )
# Create managers
resources_manager = DorisResourcesManager(_worker_connection_manager)
@@ -149,15 +160,50 @@ async def initialize_worker():
raise
async def health_check(request):
- """Health check endpoint that shows worker PID"""
- return JSONResponse({
- "status": "healthy",
- "service": "doris-mcp-server",
- "worker_pid": os.getpid(),
- "worker_mode": "multi-process-full-mcp",
- "mcp_initialized": _worker_initialized,
- "version": __version__,
- })
+ """Backward-compatible liveness endpoint."""
+ return JSONResponse(
+ liveness_payload(
+ service="doris-mcp-server",
+ version=__version__,
+ legacy=True,
+ details={
+ "worker_pid": os.getpid(),
+ "worker_mode": "multi-process-full-mcp",
+ "mcp_initialized": _worker_initialized,
+ },
+ )
+ )
+
+
+async def live_check(request):
+ """Database-independent liveness endpoint."""
+ return JSONResponse(
+ liveness_payload(
+ service="doris-mcp-server",
+ version=__version__,
+ details={
+ "worker_pid": os.getpid(),
+ "worker_mode": "multi-process-full-mcp",
+ "mcp_initialized": _worker_initialized,
+ },
+ )
+ )
+
+
+async def readiness_check(request):
+ """Bounded readiness endpoint for this worker."""
+ payload, status_code = await readiness_payload(
+ _worker_connection_manager,
+ service="doris-mcp-server",
+ version=__version__,
+ initialized=_worker_initialized,
+ details={
+ "worker_pid": os.getpid(),
+ "worker_mode": "multi-process-full-mcp",
+ "mcp_initialized": _worker_initialized,
+ },
+ )
+ return JSONResponse(payload, status_code=status_code)
# OAuth and Token handlers (initialize after worker setup)
_oauth_handlers = None
@@ -304,6 +350,8 @@ async def root_info(request):
"version": __version__,
"endpoints": {
"health": "/health",
+ "live": "/live",
+ "ready": "/ready",
"mcp": "/mcp"
}
})
@@ -399,6 +447,8 @@ basic_app = Starlette(
routes=[
Route("/", root_info, methods=["GET"]),
Route("/health", health_check, methods=["GET"]),
+ Route("/live", live_check, methods=["GET"]),
+ Route("/ready", readiness_check, methods=["GET"]),
# OAuth endpoints
Route("/auth/login", oauth_login, methods=["GET"]),
Route("/auth/callback", oauth_callback, methods=["GET"]),
diff --git a/doris_mcp_server/utils/db.py b/doris_mcp_server/utils/db.py
index ac06b4b..5e1f37a 100644
--- a/doris_mcp_server/utils/db.py
+++ b/doris_mcp_server/utils/db.py
@@ -1507,7 +1507,32 @@ class DorisConnectionManager:
except Exception as e:
self.logger.error(f"Database connectivity test failed: {e}")
return False
-
+
+ async def check_readiness(self, *, timeout_seconds: float = 2.0) -> bool:
+ """Run a bounded, side-effect-free Doris readiness probe."""
+ if not self._has_valid_global_config():
+ return False
+ try:
+ timeout = float(timeout_seconds)
+ except (TypeError, ValueError):
+ timeout = 2.0
+ timeout = min(max(timeout, 0.05), 5.0)
+ try:
+ await asyncio.wait_for(
+ self._test_basic_connectivity(),
+ timeout=timeout,
+ )
+ return True
+ except TimeoutError:
+ self.logger.debug(
+ "Doris readiness probe timed out after %s seconds",
+ timeout,
+ )
+ return False
+ except Exception as exc:
+ self.logger.debug("Doris readiness probe failed: %s", exc)
+ return False
+
async def _test_basic_connectivity(self) -> None:
"""
Test basic database connectivity without connection pool
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index 3f3dc01..0c70a04 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -247,21 +247,22 @@ async def _wait_for_http_server(
pytrace=False,
)
try:
- response = await client.get(f"http://127.0.0.1:{port}/health";)
+ response = await client.get(f"http://127.0.0.1:{port}/live";)
if response.status_code == 200:
assert response.json()["version"] == __version__
+ assert response.json()["status"] == "alive"
return
except httpx.HTTPError:
pass
await asyncio.sleep(0.1)
pytest.fail(
- "HTTP MCP process did not become healthy:\n" +
_read_process_log(log_file),
+ "HTTP MCP process did not become live:\n" +
_read_process_log(log_file),
pytrace=False,
)
@asynccontextmanager
-async def _http_client(environment: dict[str, str]) -> AsyncIterator[Client]:
+async def _http_process(environment: dict[str, str]) -> AsyncIterator[str]:
port = _free_tcp_port()
with tempfile.TemporaryFile() as log_file:
process = await asyncio.create_subprocess_exec(
@@ -281,15 +282,21 @@ async def _http_client(environment: dict[str, str]) ->
AsyncIterator[Client]:
)
try:
await _wait_for_http_server(process, port, log_file)
- async with Client(
- f"http://127.0.0.1:{port}/mcp";,
- read_timeout_seconds=15,
- ) as client:
- yield client
+ yield f"http://127.0.0.1:{port}";
finally:
await _stop_process(process)
+@asynccontextmanager
+async def _http_client(environment: dict[str, str]) -> AsyncIterator[Client]:
+ async with _http_process(environment) as base_url:
+ async with Client(
+ f"{base_url}/mcp",
+ read_timeout_seconds=15,
+ ) as client:
+ yield client
+
+
@asynccontextmanager
async def _stdio_client(environment: dict[str, str]) -> AsyncIterator[Client]:
parameters = StdioServerParameters(
@@ -346,6 +353,32 @@ async def _exec_query(
return result, result.structured_content
+async def test_real_doris_liveness_and_readiness(
+ doris_sandbox: DorisSandbox,
+) -> None:
+ environment = _server_environment(
+ doris_sandbox.settings,
+ user=doris_sandbox.settings.user,
+ password=doris_sandbox.settings.password,
+ )
+ async with _http_process(environment) as base_url:
+ async with httpx.AsyncClient(timeout=5) as client:
+ health = await client.get(f"{base_url}/health")
+ live = await client.get(f"{base_url}/live")
+ ready = await client.get(f"{base_url}/ready")
+
+ assert health.status_code == 200
+ assert health.json()["status"] == "healthy"
+ assert live.status_code == 200
+ assert live.json()["status"] == "alive"
+ assert ready.status_code == 200
+ assert ready.json()["status"] == "ready"
+ assert ready.json()["checks"] == {
+ "service": "ready",
+ "doris": "ready",
+ }
+
+
@pytest.mark.parametrize("transport", ["http", "stdio"])
async def test_real_doris_read_write_permission_timeout_and_recovery(
transport: str,
diff --git a/test/protocol/test_health_endpoints.py
b/test/protocol/test_health_endpoints.py
new file mode 100644
index 0000000..84c0114
--- /dev/null
+++ b/test/protocol/test_health_endpoints.py
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Liveness and readiness regression tests."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from mcp import Client
+
+from doris_mcp_server import __version__
+from doris_mcp_server.health import liveness_payload, readiness_payload
+from doris_mcp_server.utils.config import DorisConfig
+from doris_mcp_server.utils.db import DorisConnectionManager
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _free_tcp_port() -> int:
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ return int(sock.getsockname()[1])
+
+
+async def _stop_process(process: asyncio.subprocess.Process) -> None:
+ if process.returncode is not None:
+ return
+ process.terminate()
+ try:
+ await asyncio.wait_for(process.wait(), timeout=10)
+ except TimeoutError:
+ process.kill()
+ await process.wait()
+
+
+async def _wait_for_live(
+ process: asyncio.subprocess.Process,
+ port: int,
+ log_file,
+) -> None:
+ deadline = time.monotonic() + 15
+ async with httpx.AsyncClient(timeout=0.5) as client:
+ while time.monotonic() < deadline:
+ if process.returncode is not None:
+ log_file.seek(0)
+ pytest.fail(
+ "HTTP process exited before liveness became available:\n"
+ + log_file.read().decode("utf-8", errors="replace"),
+ pytrace=False,
+ )
+ try:
+ response = await client.get(f"http://127.0.0.1:{port}/live";)
+ if response.status_code == 200:
+ assert response.json()["status"] == "alive"
+ return
+ except httpx.HTTPError:
+ pass
+ await asyncio.sleep(0.1)
+ log_file.seek(0)
+ pytest.fail(
+ "HTTP process did not expose liveness:\n"
+ + log_file.read().decode("utf-8", errors="replace"),
+ pytrace=False,
+ )
+
+
+def test_liveness_payload_is_database_independent() -> None:
+ assert liveness_payload(
+ service="doris-mcp-server",
+ version=__version__,
+ ) == {
+ "status": "alive",
+ "service": "doris-mcp-server",
+ "version": __version__,
+ "checks": {"process": "alive"},
+ }
+
+
[email protected]
+async def test_readiness_recovers_without_changing_liveness() -> None:
+ manager = SimpleNamespace(
+ check_readiness=AsyncMock(side_effect=[False, True]),
+ )
+
+ unavailable, unavailable_status = await readiness_payload(
+ manager,
+ service="doris-mcp-server",
+ version=__version__,
+ )
+ recovered, recovered_status = await readiness_payload(
+ manager,
+ service="doris-mcp-server",
+ version=__version__,
+ )
+
+ assert unavailable_status == 503
+ assert unavailable["status"] == "not_ready"
+ assert unavailable["checks"]["doris"] == "not_ready"
+ assert recovered_status == 200
+ assert recovered["status"] == "ready"
+ assert recovered["checks"]["doris"] == "ready"
+ assert (
+ liveness_payload(
+ service="doris-mcp-server",
+ version=__version__,
+ )["status"]
+ == "alive"
+ )
+
+
[email protected]
+async def test_readiness_has_a_hard_timeout() -> None:
+ async def never_finishes(*, timeout_seconds: float) -> bool:
+ del timeout_seconds
+ await asyncio.sleep(60)
+ return True
+
+ manager = SimpleNamespace(check_readiness=never_finishes)
+ started = time.monotonic()
+ payload, status_code = await readiness_payload(
+ manager,
+ service="doris-mcp-server",
+ version=__version__,
+ timeout_seconds=0.01,
+ )
+
+ assert time.monotonic() - started < 0.5
+ assert status_code == 503
+ assert payload["status"] == "not_ready"
+
+
[email protected]
+async def test_connection_readiness_uses_a_direct_probe() -> None:
+ config = DorisConfig()
+ manager = DorisConnectionManager(config)
+ manager._test_basic_connectivity = AsyncMock(return_value=None)
+
+ assert await manager.check_readiness(timeout_seconds=60) is True
+ manager._test_basic_connectivity.assert_awaited_once_with()
+
+
[email protected]
+async def test_connection_readiness_direct_probe_times_out() -> None:
+ config = DorisConfig()
+ manager = DorisConnectionManager(config)
+
+ async def slow_connectivity() -> None:
+ await asyncio.sleep(60)
+
+ manager._test_basic_connectivity = slow_connectivity
+ started = time.monotonic()
+ assert await manager.check_readiness(timeout_seconds=0.01) is False
+ assert time.monotonic() - started < 0.5
+
+
[email protected]
[email protected]("workers", [1, 2])
+async def test_http_process_stays_live_when_doris_is_unavailable(
+ workers: int,
+) -> None:
+ server_port = _free_tcp_port()
+ unavailable_doris_port = _free_tcp_port()
+ environment = os.environ.copy()
+ environment.update(
+ {
+ "DORIS_HOST": "127.0.0.1",
+ "DORIS_PORT": str(unavailable_doris_port),
+ "DORIS_USER": "root",
+ "DORIS_PASSWORD": "",
+ "DORIS_DATABASE": "information_schema",
+ "DORIS_CONNECTION_TIMEOUT": "1",
+ "ENABLE_TOKEN_AUTH": "false",
+ "ENABLE_JWT_AUTH": "false",
+ "ENABLE_OAUTH_AUTH": "false",
+ "ENABLE_DORIS_OAUTH_AUTH": "false",
+ "LOG_LEVEL": "ERROR",
+ "PYTHONUNBUFFERED": "1",
+ }
+ )
+
+ with tempfile.TemporaryFile() as log_file:
+ process = await asyncio.create_subprocess_exec(
+ sys.executable,
+ "-m",
+ "doris_mcp_server",
+ "--transport",
+ "http",
+ "--host",
+ "127.0.0.1",
+ "--port",
+ str(server_port),
+ "--workers",
+ str(workers),
+ cwd=PROJECT_ROOT,
+ env=environment,
+ stdout=log_file,
+ stderr=subprocess.STDOUT,
+ )
+ try:
+ await _wait_for_live(process, server_port, log_file)
+ async with httpx.AsyncClient(timeout=3) as client:
+ health = await
client.get(f"http://127.0.0.1:{server_port}/health";)
+ live = await client.get(f"http://127.0.0.1:{server_port}/live";)
+ started = time.monotonic()
+ ready = await
client.get(f"http://127.0.0.1:{server_port}/ready";)
+ readiness_elapsed = time.monotonic() - started
+
+ assert health.status_code == 200
+ assert health.json()["status"] == "healthy"
+ assert live.status_code == 200
+ assert live.json()["status"] == "alive"
+ assert ready.status_code == 503
+ assert ready.json()["status"] == "not_ready"
+ assert readiness_elapsed < 3
+
+ async with Client(
+ f"http://127.0.0.1:{server_port}/mcp";,
+ read_timeout_seconds=10,
+ ) as client:
+ tools = await client.list_tools(cache_mode="bypass")
+ assert tools.tools
+
+ assert process.returncode is None
+ finally:
+ await _stop_process(process)
diff --git a/test/protocol/test_multiworker_config.py
b/test/protocol/test_multiworker_config.py
index fdc5810..cccd204 100644
--- a/test/protocol/test_multiworker_config.py
+++ b/test/protocol/test_multiworker_config.py
@@ -1,10 +1,17 @@
import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
import pytest
from doris_mcp_server import __version__
from doris_mcp_server.main import _multiworker_environment
-from doris_mcp_server.multiworker_app import health_check, root_info
+from doris_mcp_server.multiworker_app import (
+ health_check,
+ live_check,
+ readiness_check,
+ root_info,
+)
from doris_mcp_server.utils.config import DorisConfig
@@ -44,9 +51,45 @@ def
test_multiworker_environment_preserves_resolved_parent_config(monkeypatch):
@pytest.mark.asyncio
async def test_multiworker_http_identity_reports_product_version():
- for handler in (health_check, root_info):
+ for handler in (health_check, live_check, root_info):
response = await handler(None)
payload = json.loads(response.body)
assert payload["service"] == "doris-mcp-server"
assert payload["version"] == __version__
assert "mcp_version" not in payload
+
+
[email protected]
+async def test_multiworker_readiness_requires_initialization(monkeypatch):
+ from doris_mcp_server import multiworker_app
+
+ readiness_probe = AsyncMock(
+ side_effect=AssertionError("uninitialized worker must not probe Doris")
+ )
+ manager = SimpleNamespace(check_readiness=readiness_probe)
+ monkeypatch.setattr(multiworker_app, "_worker_initialized", False)
+ monkeypatch.setattr(multiworker_app, "_worker_connection_manager", manager)
+
+ response = await readiness_check(None)
+ payload = json.loads(response.body)
+ assert response.status_code == 503
+ assert payload["status"] == "not_ready"
+ assert payload["checks"]["service"] == "not_ready"
+ readiness_probe.assert_not_awaited()
+
+
[email protected]
+async def test_multiworker_readiness_uses_worker_database_probe(monkeypatch):
+ from doris_mcp_server import multiworker_app
+
+ readiness_probe = AsyncMock(return_value=True)
+ manager = SimpleNamespace(check_readiness=readiness_probe)
+ monkeypatch.setattr(multiworker_app, "_worker_initialized", True)
+ monkeypatch.setattr(multiworker_app, "_worker_connection_manager", manager)
+
+ response = await readiness_check(None)
+ payload = json.loads(response.body)
+ assert response.status_code == 200
+ assert payload["status"] == "ready"
+ assert payload["checks"]["doris"] == "ready"
+ readiness_probe.assert_awaited_once_with(timeout_seconds=2.0)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
