136 lines
6.5 KiB
Python
136 lines
6.5 KiB
Python
"""Dedicated single-process online execution entrypoint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hmac
|
|
import os
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
|
|
from starlette.responses import JSONResponse
|
|
|
|
from app.gateway.app import create_app
|
|
from app.gateway.deps import langgraph_runtime
|
|
from app.gateway.runtime_executor import configure_runtime_executor
|
|
from deerflow.config import get_app_config
|
|
|
|
from .admission import Admission
|
|
from .manager import WorkerRunManager
|
|
from .preflight import shared_storage_configured, validate_sqlite_mount
|
|
from .proxy import THREAD_PATH
|
|
from .refresh import SharedSkillRefresh
|
|
|
|
|
|
async def _measure_worker_event_loop_lag(app):
|
|
loop = asyncio.get_running_loop()
|
|
while True:
|
|
if not getattr(app.state, "worker_diagnostics_enabled", False):
|
|
app.state.worker_event_loop_lag_ms = 0.0
|
|
await asyncio.sleep(1)
|
|
continue
|
|
started = loop.time()
|
|
await asyncio.sleep(1)
|
|
app.state.worker_event_loop_lag_ms = round(max(0.0, loop.time() - started - 1) * 1000, 2)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def worker_lifespan(app):
|
|
from .revision import code_revision
|
|
app.state.worker_process_started = time.time()
|
|
app.state.worker_runtime_revision = code_revision()
|
|
config = get_app_config()
|
|
from .runtime import verify_storage_dependencies
|
|
|
|
verify_storage_dependencies(config)
|
|
if not shared_storage_configured(config):
|
|
raise RuntimeError("Managed workers require persistent PostgreSQL or SQLite chat storage")
|
|
if "AioSandboxProvider" in str(getattr(config.sandbox, "use", "")):
|
|
raise RuntimeError("Managed online workers cannot launch Docker sandboxes")
|
|
validate_sqlite_mount(config)
|
|
# Shared event records follow conversations across drained containers.
|
|
config = config.model_copy(
|
|
update={
|
|
"run_events": config.run_events.model_copy(update={"backend": "db"}),
|
|
"database": config.database.model_copy(update={"pool_size": int(os.environ.get("DEER_FLOW_WORKER_DB_POOL_SIZE", "4")), "max_overflow": 0}),
|
|
}
|
|
)
|
|
app.state.config = config
|
|
app.state.worker_skill_refresh = SharedSkillRefresh()
|
|
app.state.worker_deferred_admission = True
|
|
app.state.worker_admission = Admission(int(os.environ.get("DEER_FLOW_WORKER_CONCURRENCY", "4")), int(os.environ.get("DEER_FLOW_WORKER_QUEUE_LIMIT", "128")), float(os.environ.get("DEER_FLOW_WORKER_QUEUE_TIMEOUT", "60")))
|
|
# Each execution container keeps its own bounded rolling request metrics;
|
|
# the controller reads a compact summary with the normal status probe.
|
|
from .telemetry import RequestRecorder
|
|
|
|
app.state.request_recorder = RequestRecorder()
|
|
app.state.worker_diagnostics_enabled = False
|
|
app.state.worker_event_loop_lag_ms = 0.0
|
|
await configure_runtime_executor(config.runtime_performance)
|
|
async with langgraph_runtime(app, background=False):
|
|
from langgraph.checkpoint.memory import InMemorySaver
|
|
from langgraph.store.memory import InMemoryStore
|
|
|
|
if isinstance(app.state.checkpointer, InMemorySaver) or isinstance(app.state.store, InMemoryStore):
|
|
raise RuntimeError("Managed worker refuses an in-memory persistence fallback")
|
|
close_manager = getattr(app.state.run_manager, "aclose", None)
|
|
if close_manager is not None:
|
|
await close_manager()
|
|
app.state.run_manager = WorkerRunManager(store=app.state.run_store)
|
|
retention = getattr(config, "runtime_contention", None)
|
|
if retention is not None and hasattr(app.state.run_manager, "start_retention"):
|
|
app.state.run_manager.start_retention(retention_seconds=retention.completed_run_retention_seconds, interval_seconds=retention.run_sweep_seconds)
|
|
lag_task = asyncio.create_task(_measure_worker_event_loop_lag(app))
|
|
try:
|
|
yield
|
|
finally:
|
|
lag_task.cancel()
|
|
await asyncio.gather(lag_task, return_exceptions=True)
|
|
|
|
|
|
class InternalOnly:
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
if scope["type"] != "http":
|
|
return await self.app(scope, receive, send)
|
|
expected = os.environ.get("DEER_FLOW_WORKER_TOKEN", "")
|
|
token = dict(scope["headers"]).get(b"x-deerflow-worker-token", b"").decode()
|
|
if not expected or not hmac.compare_digest(token, expected):
|
|
return await JSONResponse({"detail": "Forbidden"}, 403)(scope, receive, send)
|
|
if scope["path"] == "/internal/worker-status":
|
|
gate = scope["app"].state.worker_admission
|
|
recorder = getattr(scope["app"].state, "request_recorder", None)
|
|
diagnostics_enabled = dict(scope["headers"]).get(b"x-deerflow-diagnostics-enabled") == b"true"
|
|
scope["app"].state.worker_diagnostics_enabled = diagnostics_enabled
|
|
if recorder is not None:
|
|
recorder.enabled = diagnostics_enabled
|
|
http = recorder.worker_status() if recorder is not None else None
|
|
return await JSONResponse({"ready": True, "active": gate.active + gate.waiting, "running": gate.active, "waiting": gate.waiting, "capacity": gate.capacity,
|
|
"process_started_at": scope["app"].state.worker_process_started,
|
|
"runtime_revision": scope["app"].state.worker_runtime_revision,
|
|
"event_loop_lag_ms": getattr(scope["app"].state, "worker_event_loop_lag_ms", None),
|
|
"diagnostics_enabled": diagnostics_enabled,
|
|
"http": http})(scope, receive, send)
|
|
active_prefix = "/internal/worker-thread-active/"
|
|
if scope["path"].startswith(active_prefix):
|
|
from urllib.parse import unquote
|
|
|
|
thread_id = unquote(scope["path"][len(active_prefix):])
|
|
active = scope["app"].state.run_manager.has_active_thread(thread_id)
|
|
return await JSONResponse({"active": active})(scope, receive, send)
|
|
match = THREAD_PATH.fullmatch(scope["path"])
|
|
if not match:
|
|
return await JSONResponse({"detail": "Worker endpoint not available"}, 404)(scope, receive, send)
|
|
scope.setdefault("state", {})["worker_node"] = os.environ.get("HOSTNAME", "worker")
|
|
from deerflow.runtime.thread_paths import aresolve_path_user_id
|
|
|
|
await aresolve_path_user_id(match[1])
|
|
await self.app(scope, receive, send)
|
|
|
|
|
|
app = create_app()
|
|
app.router.lifespan_context = worker_lifespan
|
|
app.add_middleware(InternalOnly)
|