From 6ff2127de7aba9358781f2d022b150b05e06833a Mon Sep 17 00:00:00 2001
From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com>
Date: Fri, 7 Aug 2026 19:19:25 +0800
Subject: [PATCH] update: add logger info

---
 schedule/src/schedule/execution.py       | 49 ++++++++++++++++++-
 schedule/src/schedule/main.py            |  6 +++
 schedule/src/schedule/notebook_runner.py | 15 ++++++
 schedule/src/schedule/orchestrator.py    | 60 +++++++++++++++++++++++-
 schedule/src/schedule/scheduler.py       | 17 +++++++
 schedule/src/schedule/service.py         | 11 +++++
 schedule/src/schedule/storage_client.py  | 21 +++++++++
 schedule/src/schedule/worker.py          | 42 ++++++++++++++++-
 8 files changed, 217 insertions(+), 4 deletions(-)

diff --git a/schedule/src/schedule/execution.py b/schedule/src/schedule/execution.py
index 2e342a6..d41a570 100644
--- a/schedule/src/schedule/execution.py
+++ b/schedule/src/schedule/execution.py
@@ -7,6 +7,8 @@ import tempfile
 from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 
+from loguru import logger
+
 
 MAX_LOG_BYTES = 4 * 1024 * 1024
 
@@ -39,6 +41,12 @@ async def _execute_notebook(
     timeout_seconds: int,
 ) -> ExecutionResult:
     output = artifact.with_name(f"executed-{artifact_name}")
+    logger.debug(
+        "notebook exec start: artifact={} timeout={}s args={}",
+        artifact_name,
+        timeout_seconds,
+        len(arguments),
+    )
     process = await asyncio.create_subprocess_exec(
         sys.executable,
         "-m",
@@ -67,6 +75,11 @@ async def _execute_notebook(
         exit_code = None
         error_code = "NODE_TIMEOUT"
         error_message = f"notebook exceeded timeout of {timeout_seconds} seconds"
+        logger.warning(
+            "notebook exec timed out: artifact={} timeout={}s",
+            artifact_name,
+            timeout_seconds,
+        )
     else:
         exit_code = process.returncode
         status = (
@@ -90,7 +103,7 @@ async def _execute_notebook(
             if status == "timed_out"
             else "Notebook execution failed; see node log"
         )
-    return ExecutionResult(
+    result = ExecutionResult(
         status=status,
         exit_code=exit_code,
         logs=_limited_log(stdout.decode("utf-8", errors="replace")),
@@ -100,6 +113,13 @@ async def _execute_notebook(
         error_code=error_code,
         error_message=error_message,
     )
+    logger.info(
+        "notebook exec done: artifact={} status={} exit_code={}",
+        artifact_name,
+        result.status,
+        result.exit_code,
+    )
+    return result
 
 
 async def _execute_python(
@@ -108,6 +128,12 @@ async def _execute_python(
     arguments: list[str],
     timeout_seconds: int,
 ) -> ExecutionResult:
+    logger.debug(
+        "python exec start: artifact={} timeout={}s args={}",
+        artifact.name,
+        timeout_seconds,
+        len(arguments),
+    )
     process = await asyncio.create_subprocess_exec(
         sys.executable,
         str(artifact),
@@ -129,6 +155,11 @@ async def _execute_python(
             {"status": "timed_out", "exit_code": None, "message": message},
             ensure_ascii=False,
         ).encode("utf-8")
+        logger.warning(
+            "python exec timed out: artifact={} timeout={}s",
+            artifact.name,
+            timeout_seconds,
+        )
         return ExecutionResult(
             status="timed_out",
             exit_code=None,
@@ -148,7 +179,7 @@ async def _execute_python(
         {"status": status, "exit_code": exit_code},
         ensure_ascii=False,
     ).encode("utf-8")
-    return ExecutionResult(
+    result = ExecutionResult(
         status=status,
         exit_code=exit_code,
         logs=_limited_log(stdout.decode("utf-8", errors="replace")),
@@ -158,6 +189,13 @@ async def _execute_python(
         error_code=None if status == "succeeded" else "PROCESS_EXIT_NONZERO",
         error_message=message,
     )
+    logger.info(
+        "python exec done: artifact={} status={} exit_code={}",
+        artifact.name,
+        result.status,
+        result.exit_code,
+    )
+    return result
 
 
 async def execute_artifact(
@@ -170,6 +208,12 @@ async def execute_artifact(
     arguments: list[str],
     timeout_seconds: int,
 ) -> ExecutionResult:
+    logger.debug(
+        "execute_artifact: run={} node={} script_type={}",
+        run_id[-12:],
+        node_run_id[-12:],
+        script_type,
+    )
     # Stage the artifact under Python's system temp dir (cleaned on context
     # exit). No local-FS volume assumption; the bytes only live for the
     # duration of the subprocess.
@@ -193,4 +237,5 @@ async def execute_artifact(
                 arguments=arguments,
                 timeout_seconds=timeout_seconds,
             )
+    logger.error("unsupported script_type: {}", script_type)
     raise ValueError(f"unsupported script_type: {script_type}")
diff --git a/schedule/src/schedule/main.py b/schedule/src/schedule/main.py
index c9553bb..b4e174a 100644
--- a/schedule/src/schedule/main.py
+++ b/schedule/src/schedule/main.py
@@ -3,6 +3,8 @@ from __future__ import annotations
 from contextlib import asynccontextmanager
 from typing import Any, AsyncIterator
 
+from loguru import logger
+
 from common.config import settings
 from common.db import create_database_engine, create_session_factory
 from common.service_app import create_service_app
@@ -16,6 +18,7 @@ from schedule.storage_client import SchedulerStorageClient
 
 @asynccontextmanager
 async def lifespan(app: Any) -> AsyncIterator[None]:
+    logger.info("scheduler service starting: service_name={}", settings.service_name)
     engine = create_database_engine(settings.database_url)
     session_factory = create_session_factory(engine)
     storage_http_client = build_storage_http_client()
@@ -28,12 +31,15 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     )
     app.state.scheduler_service = service
     await service.start()
+    logger.info("scheduler service started: cron + orchestrator loops running")
     try:
         yield
     finally:
+        logger.info("scheduler service shutting down")
         await service.close()
         await storage_http_client.aclose()
         await engine.dispose()
+        logger.info("scheduler service stopped")
 
 
 app = create_service_app(
diff --git a/schedule/src/schedule/notebook_runner.py b/schedule/src/schedule/notebook_runner.py
index 7453519..5f86e01 100644
--- a/schedule/src/schedule/notebook_runner.py
+++ b/schedule/src/schedule/notebook_runner.py
@@ -6,6 +6,8 @@ import sys
 import traceback
 from pathlib import Path
 
+from loguru import logger
+
 import nbformat
 from nbclient import NotebookClient
 
@@ -48,6 +50,12 @@ def main() -> None:
     ):
         raise ValueError("arguments-json must contain an array of strings")
 
+    logger.info(
+        "notebook runner start: input={} timeout={}s args={}",
+        source.name,
+        args.timeout,
+        len(arguments),
+    )
     notebook = nbformat.read(source, as_version=4)
     if arguments:
         notebook.cells.insert(
@@ -66,17 +74,24 @@ def main() -> None:
             kernel_name="python3",
             allow_errors=False,
         )
+        logger.debug(
+            "notebook client created: kernel=python3 timeout={}s",
+            max(1, args.timeout),
+        )
         # No explicit cwd — the kernel inherits the parent's cwd, which the
         # scheduler sets to the staged artifact directory. Keeping it here
         # avoids any "cwd must exist" requirement on the host.
         client.execute()
+        logger.info("notebook client execute done: input={}", source.name)
     except Exception as exc:
+        logger.exception("notebook execute failed: input={}", source.name)
         traceback.print_exc()
         exit_code = 124 if "timeout" in type(exc).__name__.lower() else 1
     finally:
         output.parent.mkdir(parents=True, exist_ok=True)
         nbformat.write(notebook, output)
         emit_outputs(notebook)
+        logger.debug("notebook output written: {}", output)
     raise SystemExit(exit_code)
 
 
diff --git a/schedule/src/schedule/orchestrator.py b/schedule/src/schedule/orchestrator.py
index ec69f34..2350f2a 100644
--- a/schedule/src/schedule/orchestrator.py
+++ b/schedule/src/schedule/orchestrator.py
@@ -125,6 +125,8 @@ class DispatchOrchestrator:
         while True:
             try:
                 processed = await self.process_pending_events(limit=20)
+                if processed:
+                    logger.debug("database event loop processed {} events", processed)
                 if not processed:
                     await asyncio.sleep(0.25)
             except asyncio.CancelledError:
@@ -137,6 +139,8 @@ class DispatchOrchestrator:
         while True:
             try:
                 claimed = await self._claim_execution_events(limit=10)
+                if claimed:
+                    logger.debug("execution loop claimed {} events", claimed)
                 if not claimed:
                     await asyncio.sleep(0.25)
             except asyncio.CancelledError:
@@ -203,6 +207,7 @@ class DispatchOrchestrator:
             )
             self._exec_tasks.add(task)
             task.add_done_callback(self._exec_tasks.discard)
+        logger.debug("claimed {} node execute events", len(claimed))
         return len(claimed)
 
     async def _run_node_execute(
@@ -210,6 +215,7 @@ class DispatchOrchestrator:
         envelope: dict[str, Any],
         message_id: str,
     ) -> None:
+        logger.debug("node execute start: event_id={}", envelope["event_id"][-12:])
         async with self._exec_semaphore:
             exc: Exception | None = None
             try:
@@ -236,15 +242,27 @@ class DispatchOrchestrator:
                 item.event_status = "published"
                 item.published_at = utcnow()
                 item.last_error = None
+                logger.info("node execute success: event_id={}", event_id[-12:])
             else:
                 item.retry_count += 1
                 item.last_error = str(exc)[:2000]
                 if item.retry_count >= 5:
                     item.event_status = "failed"
+                    logger.warning(
+                        "node execute exhausted retries: event_id={} retry_count={}",
+                        event_id[-12:],
+                        item.retry_count,
+                    )
                 else:
                     item.available_at = utcnow() + timedelta(
                         seconds=min(30, 2 ** item.retry_count),
                     )
+                    logger.warning(
+                        "node execute retry scheduled: event_id={} retry_count={} delay={}s",
+                        event_id[-12:],
+                        item.retry_count,
+                        min(30, 2 ** item.retry_count),
+                    )
 
     async def process_pending_events(
         self,
@@ -274,6 +292,7 @@ class DispatchOrchestrator:
                         OutboxEvents.aggregate_id == aggregate_id
                     )
                 events = list((await session.scalars(statement)).all())
+                logger.debug("processed {} outbox events", len(events))
                 for item in events:
                     try:
                         await self._process_outbox_event(item)
@@ -356,6 +375,11 @@ class DispatchOrchestrator:
         if event.get("event_type") != SCHEDULE_RUN_REQUESTED_EVENT:
             raise ValueError("unexpected event type")
         payload = event["payload"]
+        logger.info(
+            "run requested: run={} trace={}",
+            payload["run_id"][-12:],
+            event["trace_id"][-12:],
+        )
         async with session_scope(self.session_factory) as session:
             inbox, should_process = await self._start_inbox(
                 session,
@@ -422,6 +446,12 @@ class DispatchOrchestrator:
             if delay_seconds
             else utcnow()
         )
+        logger.info(
+            "dispatch node: run={} node={} attempt={}",
+            run.run_id[-12:],
+            node["node_id"][-12:],
+            attempt_no,
+        )
         await add_outbox_event(
             session,
             event_type=NODE_EXECUTE_EVENT,
@@ -473,7 +503,15 @@ class DispatchOrchestrator:
             if node["node_id"] not in target_node_ids
         ]
         max_concurrency = max(1, int(snapshot.get("max_concurrency", 1)))
-        for node in roots[:max_concurrency]:
+        selected_roots = roots[:max_concurrency]
+        if selected_roots:
+            logger.info(
+                "bootstrap roots: run={} roots_count={} max_concurrency={}",
+                run.run_id[-12:],
+                len(selected_roots),
+                max_concurrency,
+            )
+        for node in selected_roots:
             await self._dispatch_node(
                 session,
                 run=run,
@@ -599,6 +637,12 @@ class DispatchOrchestrator:
                     session.add(skipped)
                     latest[node_id] = skipped
                     changed = True
+                    logger.debug(
+                        "skipped node: run={} node={} reason={}",
+                        run.run_id[-12:],
+                        node_id[-12:],
+                        "stop_policy" if stop_all else "parents_blocked",
+                    )
                 elif (
                     # Dispatch only when every parent has actually
                     # run to completion successfully. A None parent
@@ -663,6 +707,14 @@ class DispatchOrchestrator:
                 succeeded = all(
                     status == "succeeded" for status in statuses
                 )
+            logger.info(
+                "run finalized: run={} status={} failure_policy={} any_failure={} any_success={}",
+                run.run_id[-12:],
+                "succeeded" if succeeded else "failed",
+                failure_policy,
+                any_real_failure,
+                any_success,
+            )
             run.run_status = "succeeded" if succeeded else "failed"
             run.error_code = None if succeeded else "SCHEDULE_NODE_FAILED"
             run.error_message = (
@@ -686,6 +738,12 @@ class DispatchOrchestrator:
         if event.get("event_type") != NODE_FINISHED_EVENT:
             raise ValueError("unexpected event type")
         payload = event["payload"]
+        logger.info(
+            "node finished event: run={} node={} status={}",
+            payload["run_id"][-12:],
+            payload["node_id"][-12:],
+            payload.get("node_status"),
+        )
         async with session_scope(self.session_factory) as session:
             inbox, should_process = await self._start_inbox(
                 session,
diff --git a/schedule/src/schedule/scheduler.py b/schedule/src/schedule/scheduler.py
index e97adcc..230ed8c 100644
--- a/schedule/src/schedule/scheduler.py
+++ b/schedule/src/schedule/scheduler.py
@@ -41,6 +41,7 @@ _ACTIVE_TRIGGER: Callable[[str], Awaitable[None]] | None = None
 
 async def dispatch_persisted_cron(schedule_id: str) -> None:
     """Dispatch one persisted cron tick through the active service."""
+    logger.debug("dispatch persisted cron: schedule={}", schedule_id[-12:])
     callback = _ACTIVE_TRIGGER
     if callback is None:
         raise RuntimeError("cron trigger callback is not initialized")
@@ -70,6 +71,7 @@ class CronScheduler:
 
     def start(self) -> None:
         """Start APScheduler and spawn the periodic sync loop."""
+        logger.info("cron scheduler starting")
         self.scheduler.start()
         self._sync_task = asyncio.create_task(
             self._sync_loop(),
@@ -78,6 +80,7 @@ class CronScheduler:
 
     async def close(self) -> None:
         """Cancel the sync loop and shut APScheduler down."""
+        logger.info("cron scheduler closing")
         global _ACTIVE_TRIGGER
 
         if self._sync_task is not None:
@@ -107,6 +110,7 @@ class CronScheduler:
         while True:
             try:
                 await self._sync_once()
+                logger.debug("cron sync tick ok")
             except asyncio.CancelledError:
                 raise
             except Exception:
@@ -134,6 +138,8 @@ class CronScheduler:
                 ).all()
             )
             active_job_ids: set[str] = set()
+            added_count = 0
+            updated_count = 0
             for item in schedules:
                 job_id = f"schedule:{item.schedule_id}"
                 active_job_ids.add(job_id)
@@ -144,6 +150,7 @@ class CronScheduler:
                 )
                 if self.scheduler.get_job(job_id) is not None:
                     self.scheduler.reschedule_job(job_id, trigger=trigger)
+                    updated_count += 1
                 else:
                     self.scheduler.add_job(
                         dispatch_persisted_cron,
@@ -155,14 +162,24 @@ class CronScheduler:
                         max_instances=max(1, item.max_concurrency),
                         misfire_grace_time=60,
                     )
+                    added_count += 1
                 job = self.scheduler.get_job(job_id)
                 item.next_run_at = naive_utc(job.next_run_time)
+            removed_count = 0
             for job in self.scheduler.get_jobs():
                 if (
                     job.id.startswith("schedule:")
                     and job.id not in active_job_ids
                 ):
                     self.scheduler.remove_job(job.id)
+                    removed_count += 1
+            if added_count or updated_count or removed_count:
+                logger.info(
+                    "cron sync reconciled: added={} updated={} removed={}",
+                    added_count,
+                    updated_count,
+                    removed_count,
+                )
 
 
 __all__ = ["CronScheduler"]
diff --git a/schedule/src/schedule/service.py b/schedule/src/schedule/service.py
index 575270b..decf2b8 100644
--- a/schedule/src/schedule/service.py
+++ b/schedule/src/schedule/service.py
@@ -102,9 +102,15 @@ class SchedulerService:
         """
         self.cron.start()
         self.orchestrator.start()
+        logger.info(
+            "scheduler service starting: cron={} orchestrator={}",
+            type(self.cron).__name__,
+            type(self.orchestrator).__name__,
+        )
 
     async def close(self) -> None:
         """Cancel orchestrator loop, then shut APScheduler down."""
+        logger.info("scheduler service closing")
         await self.orchestrator.close()
         await self.cron.close()
 
@@ -140,6 +146,11 @@ class SchedulerService:
         idempotency_key = (
             f"cron:{schedule_id}:{now.strftime('%Y%m%d%H%M')}"
         )
+        logger.info(
+            "cron trigger: schedule={} workspace={}",
+            schedule_id[-12:],
+            workspace_id[-12:],
+        )
         try:
             async with self.session_factory() as session:
                 await create_scheduled_run(
diff --git a/schedule/src/schedule/storage_client.py b/schedule/src/schedule/storage_client.py
index 0abc7ff..25dd58d 100644
--- a/schedule/src/schedule/storage_client.py
+++ b/schedule/src/schedule/storage_client.py
@@ -15,6 +15,8 @@ from __future__ import annotations
 import base64
 from typing import Any
 
+from loguru import logger
+
 import httpx
 
 
@@ -39,6 +41,13 @@ class SchedulerStorageClient:
         we return the inner ``data`` dict (which includes
         ``storage_object_id`` and ``storage_uri``).
         """
+        logger.debug(
+            "storage create_object: workspace={} usage_type={} file={} size={}B",
+            workspace_id[-12:],
+            usage_type,
+            file_name,
+            len(content),
+        )
         response = await self._http.post(
             "/internal/v1/objects",
             json={
@@ -54,8 +63,20 @@ class SchedulerStorageClient:
                 "relative_path": None,
             },
         )
+        if response.is_error:
+            logger.warning(
+                "storage create_object HTTP error: status={} url={}",
+                response.status_code,
+                response.request.url,
+            )
         response.raise_for_status()
         body = response.json()
+        logger.info(
+            "storage create_object done: workspace={} usage_type={} storage_object_id={}",
+            workspace_id[-12:],
+            usage_type,
+            body["data"].get("storage_object_id"),
+        )
         return body["data"]
 
 
diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py
index df3ddf8..df61ad2 100644
--- a/schedule/src/schedule/worker.py
+++ b/schedule/src/schedule/worker.py
@@ -70,6 +70,12 @@ class NodeExecutor:
         payload = event["payload"]
         if not await self._set_node_running(event, message_id):
             return
+        logger.info(
+            "node execute start: node_run={} script_type={} timeout={}s",
+            payload["node_run_id"][-12:],
+            payload["script_type"],
+            payload["timeout_seconds"],
+        )
         started_at = utcnow()
         context: dict[str, Any] | None = None
         try:
@@ -193,6 +199,12 @@ class NodeExecutor:
                     },
                 )
             self._finish_inbox(inbox)
+        logger.info(
+            "node execute done: node_run={} status={} error_code={}",
+            payload["node_run_id"][-12:],
+            final_status,
+            final_error_code,
+        )
 
     async def _set_node_running(
         self,
@@ -219,6 +231,11 @@ class NodeExecutor:
             if node_run is None:
                 raise ValueError("schedule node run does not exist")
             if node_run.node_status in TERMINAL_NODE_STATES:
+                logger.debug(
+                    "node already terminal: node_run={} status={}",
+                    payload["node_run_id"][-12:],
+                    node_run.node_status,
+                )
                 self._finish_inbox(inbox)
                 return False
             if node_run.node_status == "queued":
@@ -226,6 +243,10 @@ class NodeExecutor:
                 node_run.started_at = utcnow()
                 node_run.message = "Worker 正在执行稳定版本"
                 node_run.state_version += 1
+                logger.debug(
+                    "node running: node_run={}",
+                    payload["node_run_id"][-12:],
+                )
             return True
 
     async def _execution_context(
@@ -279,7 +300,7 @@ class NodeExecutor:
             if not storage.bucket_name or not storage.object_key:
                 raise ValueError("stable version artifact location is incomplete")
             user_id = run.triggered_by or schedule.created_by
-            return {
+            context = {
                 "node_status": node_run.node_status,
                 "workspace_id": run.workspace_id,
                 "workspace_code": workspace.workspace_code,
@@ -288,6 +309,13 @@ class NodeExecutor:
                 "object_key": storage.object_key,
                 "content_hash": version.content_hash,
             }
+            logger.debug(
+                "execution context loaded: node_run={} bucket={} object_key={}",
+                payload["node_run_id"][-12:],
+                context["bucket_name"],
+                context["object_key"][-32:],
+            )
+            return context
 
     async def _fallback_execution_context(
         self,
@@ -327,6 +355,12 @@ class NodeExecutor:
         content = await self.object_store.get(object_key)
         if hashlib.sha256(content).hexdigest() != content_hash:
             raise ValueError("stable version artifact hash mismatch")
+        logger.debug(
+            "artifact downloaded: bucket={} object_key={} bytes={}",
+            bucket_name,
+            object_key[-32:],
+            len(content),
+        )
         return content
 
     async def _upload_execution_artifacts(
@@ -363,6 +397,12 @@ class NodeExecutor:
         except Exception as exc:
             upload_error = f"result upload failed: {exc}"[:2000]
             logger.exception("failed to upload node execution artifacts")
+        if log_id and result_id and not upload_error:
+            logger.info(
+                "artifacts uploaded: log_id={} result_id={}",
+                log_id[-12:],
+                result_id[-12:],
+            )
         return log_id, result_id, upload_error
 
     @staticmethod