tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: integrate feat/auth into develop

Browse Source
This commit is contained in:
Winnie
2026-08-03 17:44:00 +08:00
parent 386efb4d98 838cbfdc16
commit d7bd88335c
38 changed files with 2590 additions and 796 deletions
Download Patch File Download Diff File Expand all files Collapse all files
schedule/Dockerfile
+4 -4
View File
@@ -7,10 +7,10 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
WORKDIR /app
RUN ( \
sed -i 's/deb.debian.org/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
sed -i 's/deb.debian.org/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list 2>/dev/null || \
sed -i 's/archive.ubuntu.com/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list 2>/dev/null || \
sed -i 's/security.ubuntu.com/mirrors.cloud.aliyuncs.com/g' /etc/apt/sources.list 2>/dev/null \
sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null || \
sed -i 's/archive.ubuntu.com/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null || \
sed -i 's/security.ubuntu.com/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null \
) || ( \
sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
sed -i 's/deb.debian.org/mirrors.aliyun.com/g' /etc/apt/sources.list 2>/dev/null || \
schedule/src/schedule/main.py
+4 -4
View File
@@ -18,12 +18,12 @@ from schedule.storage_client import SchedulerStorageClient
async def lifespan(app: Any) -> AsyncIterator[None]:
engine = create_database_engine(settings.database_url)
session_factory = create_session_factory(engine)
backend_http_client = build_storage_http_client()
storage_http_client = build_storage_http_client()
service = SchedulerService(
session_factory=session_factory,
backend_http_client=backend_http_client,
storage_http_client=storage_http_client,
object_store=build_object_store(),
storage_client=SchedulerStorageClient(backend_http_client),
storage_client=SchedulerStorageClient(storage_http_client),
database_url=settings.database_url,
)
app.state.scheduler_service = service
@@ -32,7 +32,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
yield
finally:
await service.close()
await backend_http_client.aclose()
await storage_http_client.aclose()
await engine.dispose()
schedule/src/schedule/orchestrator.py
+112 -37
View File
@@ -47,20 +47,30 @@ class DispatchOrchestrator:
- ``_database_event_loop`` drains ``schedule.run.requested`` and
``job.node.finished`` events under ``dispatch_lock``. These are
short, in-line DB transactions.
- ``_execution_loop`` claims ``job.node.execute`` events, sets a
far-future ``available_at`` as a lease, then dispatches each as
- ``_execution_loop`` claims ``job.node.execute`` events whose
``available_at <= utcnow()`` and dispatches each as
``asyncio.create_task`` so the polling path is never blocked by
notebook execution. A semaphore caps concurrent notebooks.
Holds a ``dispatch_lock`` to keep two concurrent drain loops from
fighting over the same batch.
Lease semantics live on the outbox row itself, not in the claim
step. The dispatcher sets ``available_at = utcnow() + node_timeout
+ LEASE_SLACK`` when it writes the ``job.node.execute`` event, so a
process crash mid-execution lets the row re-eligible automatically
once the lease expires. A hard-coded 30-minute lease was the
original P0-2 bug: a node with ``timeout_seconds = 86_400`` would
be re-claimed at 30 minutes and run twice; a node with
``timeout_seconds = 60`` would have its lease expire 29 minutes
too early. Tieing the lease to the actual node timeout closes both
cases.
"""
# Lease window for a claimed ``job.node.execute`` event. If the
# process dies mid-execution, the row re-eligible after this many
# minutes. The handler is idempotent (it short-circuits on terminal
# node states), so safe re-execution.
EXECUTION_LEASE = timedelta(minutes=30)
# Margin added on top of ``timeout_seconds`` when writing the lease
# ``available_at``. Gives the worker time to update the row to a
# terminal state before the poll re-picks it.
LEASE_SLACK = timedelta(seconds=30)
def __init__(
self,
@@ -140,11 +150,13 @@ class DispatchOrchestrator:
) -> int:
"""Claim ``job.node.execute`` rows and dispatch them as tasks.
The claim step bumps ``available_at`` to a far-future lease so
the polling loop does not re-pick the same row while the
dispatched task is still running. Status stays ``pending``;
the dispatched task flips it to ``published`` or ``failed``
when execution completes.
Lease is owned by the row itself (the dispatcher sets
``available_at = utcnow() + node.timeout_seconds + LEASE_SLACK``
when the event is enqueued), so this method is a pure
read-and-dispatch — no DB writes in the claim step. If the
process dies before ``_run_node_execute`` finishes, the row
re-eligible once ``available_at`` falls back to now; the worker
handler is idempotent (short-circuits on terminal node state).
"""
async with session_scope(self.session_factory) as session:
statement = (
@@ -158,17 +170,18 @@ class DispatchOrchestrator:
.limit(limit)
)
events = list((await session.scalars(statement)).all())
claimed: list[tuple[dict[str, Any], str]] = []
lease_until = utcnow() + self.EXECUTION_LEASE
for item in events:
item.available_at = lease_until
envelope = {
"event_type": item.event_type,
"event_id": item.event_id,
"trace_id": item.trace_id,
"payload": item.payload_json,
}
claimed.append((envelope, f"mysql:{item.event_id}"))
claimed: list[tuple[dict[str, Any], str]] = [
(
{
"event_type": item.event_type,
"event_id": item.event_id,
"trace_id": item.trace_id,
"payload": item.payload_json,
},
f"mysql:{item.event_id}",
)
for item in events
]
for envelope, message_id in claimed:
task = asyncio.create_task(
self._run_node_execute(envelope, message_id),
@@ -379,6 +392,20 @@ class DispatchOrchestrator:
),
)
session.add(node_run)
# Lease is owned by the outbox row, not the claim step. Pick
# the later of (now, scheduled retry) and the timeout + slack,
# so a slow node isn't re-dispatched while it's still running
# but a crashed node does become eligible again after its
# timeout expires. See P0-2 in the auth refactor plan.
retry_at = (
utcnow() + timedelta(seconds=delay_seconds)
if delay_seconds
else utcnow()
)
lease_at = utcnow() + timedelta(
seconds=int(node["timeout_seconds"])
) + self.LEASE_SLACK
available_at = max(retry_at, lease_at)
await add_outbox_event(
session,
event_type="job.node.execute",
@@ -387,11 +414,7 @@ class DispatchOrchestrator:
aggregate_type="schedule_node_run",
aggregate_id=node_run.node_run_id,
idempotency_key=f"{node_run.node_run_id}:{attempt_no}",
available_at=(
utcnow() + timedelta(seconds=delay_seconds)
if delay_seconds
else None
),
available_at=available_at,
payload={
"workspace_id": run.workspace_id,
"run_id": run.run_id,
@@ -480,13 +503,29 @@ class DispatchOrchestrator:
if node_id in latest:
continue
parent_runs = [latest.get(parent) for parent in parents[node_id]]
parent_failed = any(
# Decide whether this node should be skipped. A node
# is only skipped when we know it can never run:
# * ``stop_all`` — the whole run was aborted on the
# first failure, so any not-yet-dispatched node is
# dropped;
# * all parents are terminal AND at least one
# failed — there is no remaining success path.
# If even one parent is still ``queued`` or
# ``running`` we keep waiting: under ``failure_policy
# == 'continue'`` a sibling might still succeed and
# the failed parent does not block that.
parents_terminal = all(
item is not None
and item.node_status in TERMINAL_NODE_STATES
and item.node_status != "succeeded"
for item in parent_runs
)
if stop_all or parent_failed:
any_parent_failed = any(
item is not None
and item.node_status in FAILED_NODE_STATES
for item in parent_runs
)
parents_blocked = parents_terminal and any_parent_failed
if stop_all or parents_blocked:
skipped = ScheduleNodeRuns(
node_run_id=new_ulid(),
run_id=run.run_id,
@@ -500,15 +539,24 @@ class DispatchOrchestrator:
message=(
"调度失败策略为 stop，未再启动"
if stop_all
else "上游节点未成功，已跳过"
else "上游节点全部终止且至少一个失败，已跳过"
),
)
session.add(skipped)
latest[node_id] = skipped
changed = True
elif (
all(
item is not None and item.node_status == "succeeded"
# Dispatch only when every parent has actually
# run to completion successfully. A None parent
# means the parent has not even been dispatched
# yet (e.g. upstream is still queued); the existing
# parents_blocked branch above handles the case
# where every parent is terminal but at least one
# failed.
len(parent_runs) > 0
and all(
item is not None
and item.node_status == "succeeded"
for item in parent_runs
)
and active_count < max_concurrency
@@ -531,13 +579,40 @@ class DispatchOrchestrator:
for item in latest.values()
):
now = utcnow()
succeeded = all(
item.node_status == "succeeded" for item in latest.values()
# Final run status depends on the schedule's
# ``failure_policy``. ``stop`` keeps the legacy rule — any
# non-success node fails the whole run. ``continue`` is
# more lenient: the run is a success when at least one
# root-level node succeeded and there is no remaining
# ``failed`` / ``cancelled`` / ``timed_out`` node that
# would have produced real artifacts had it run. Nodes
# marked ``skipped`` count as "decided to not run" and do
# not by themselves fail the run.
failure_policy = snapshot.get("failure_policy", "stop")
statuses = [item.node_status for item in latest.values()]
any_real_failure = any(
status in FAILED_NODE_STATES for status in statuses
)
any_success = any(
status == "succeeded" for status in statuses
)
if failure_policy == "continue":
# A run with mixed success/failure/skip outcomes is
# only "succeeded" when at least one node actually ran
# to completion and nothing hit a hard failure. A
# run where every node was skipped or failed is
# itself a failure.
succeeded = any_success and not any_real_failure
else:
succeeded = all(
status == "succeeded" for status in statuses
)
run.run_status = "succeeded" if succeeded else "failed"
run.error_code = None if succeeded else "SCHEDULE_NODE_FAILED"
run.error_message = (
None if succeeded else "one or more schedule nodes did not succeed"
None
if succeeded
else "one or more schedule nodes did not succeed"
)
run.finished_at = now
if run.started_at:
schedule/src/schedule/service.py
+52 -28
View File
@@ -6,13 +6,16 @@ Composes three single-purpose components into one bootable service:
- :class:`schedule.orchestrator.DispatchOrchestrator` — Outbox polling + DAG
- :class:`schedule.worker.NodeExecutor` — node-level execution
This module also exposes the two factory functions (``build_object_store`` /
``build_storage_http_client``) consumed by ``schedule.main`` to construct
the backing resources that flow into the facade.
This module also exposes the factory function ``build_object_store``
consumed by ``schedule.main`` to construct the RustFS S3 client.
The ``SchedulerService`` itself stays small: it wires the three components
together and implements ``trigger_schedule``, the cron post-back to
Backend that ``CronScheduler`` calls at every cron tick.
The :class:`SchedulerService` itself stays small: it wires the three
components together and implements :meth:`SchedulerService.trigger_schedule`,
the cron tick handler. The trigger writes the new ``ScheduleRuns`` row
and ``schedule.run.requested`` outbox event in a single transaction
via :func:`common.scheduler.create_scheduled_run` — no HTTP call to
the backend, so no service-to-service auth is needed (the schedule
service shares the same MySQL via the Docker network).
"""
from __future__ import annotations
@@ -25,6 +28,11 @@ import httpx
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from common.config import settings
from common.scheduler import (
SYSTEM_CRON_USER_ID,
TriggerError,
create_scheduled_run,
)
from common.ids import new_ulid
from schedule.orchestrator import DispatchOrchestrator
@@ -49,13 +57,13 @@ class SchedulerService:
self,
*,
session_factory: async_sessionmaker[AsyncSession],
backend_http_client: httpx.AsyncClient,
storage_http_client: httpx.AsyncClient,
object_store: Any,
storage_client: Any,
database_url: str,
) -> None:
self.session_factory = session_factory
self.backend_http_client = backend_http_client
self.storage_http_client = storage_http_client
self.object_store = object_store
self.storage_client = storage_client
self.database_url = database_url
@@ -99,11 +107,17 @@ class SchedulerService:
await self.cron.close()
async def trigger_schedule(self, schedule_id: str) -> None:
"""Cron tick callback: post back to Backend to register a new run.
"""Cron tick callback: write a new ``ScheduleRuns`` row + outbox event.
Backend writes the ``schedule.run.requested`` Outbox row in the same
transaction as the ``ScheduleRuns`` insert; the orchestrator's polling
loop will pick it up and start advancing the DAG.
Runs in its own session. The cron ``triggered_by`` is the
stable :data:`common.scheduler.SYSTEM_CRON_USER_ID` (the
bootstrap migration seeds the matching ``Users`` row) — we no
longer impersonate the schedule's human creator as the previous
header-based implementation did.
Idempotency key is the cron minute bucket, so re-entering the
same tick (e.g. after a brief outage) reuses the existing run
via the unique constraint on ``schedule_runs.idempotency_key``.
"""
async with self.session_factory() as session:
from sqlalchemy import select
@@ -118,26 +132,29 @@ class SchedulerService:
or item.trigger_type != "cron"
):
return
user_id = item.created_by
workspace_id = item.workspace_id
now = datetime.now(UTC)
idempotency_key = (
f"cron:{schedule_id}:{now.strftime('%Y%m%d%H%M')}"
)
response = await self.backend_http_client.post(
f"/api/v1/schedules/{schedule_id}/run",
headers={
"X-User-ID": user_id,
"X-Workspace-ID": workspace_id,
"X-Request-ID": new_ulid(),
"Idempotency-Key": idempotency_key,
},
json={"reason": "cron"},
)
if response.is_error:
raise RuntimeError(
f"backend rejected cron run: {response.status_code} "
f"{response.text[:500]}"
try:
async with self.session_factory() as session:
await create_scheduled_run(
session,
schedule_id=schedule_id,
workspace_id=workspace_id,
triggered_by_user_id=SYSTEM_CRON_USER_ID,
trigger_type="cron",
idempotency_key=idempotency_key,
trace_id=new_ulid(),
)
await session.commit()
except TriggerError as exc:
# Most likely: idempotency_key collision from a previous
# tick in the same minute — silently no-op.
LOGGER.info(
"cron trigger no-op for schedule %s: %s", schedule_id, exc,
)
async def process_pending_events(
@@ -175,7 +192,14 @@ def build_object_store() -> Any:
def build_storage_http_client() -> httpx.AsyncClient:
"""Construct the httpx client that talks to Backend's HTTP API."""
"""Construct the httpx client that talks to Backend's storage API.
The schedule service no longer needs to call any user-facing
endpoint (cron trigger writes directly to the DB now), but the
storage endpoints at ``/internal/v1/...`` still live on the
backend process and are reached via this client. Auth is not
required — the client is bound to the shared Docker network.
"""
return httpx.AsyncClient(
base_url=settings.backend_api_url,
timeout=httpx.Timeout(60.0),