worker: cover goal docker paths with prestart stage logging
PR #145 added docker_run_* lines only to _run_in_container, but goal/goal-resume/goal-resume-breakpoint call _call_runner directly (to keep full stdout for goal conclusions), so the production live test (a goal task) produced zero docker_run_ lines. Extract a _docker_call_staged helper and route all four task paths through it; error lines log the class name then re-raise so each caller's kill+failed handling is unchanged. Also fix audit attribution: audit_log has no task_id column, and worker_exec_done/error details carried no task id, so per-task queries missed them (the rows existed). All three exec details now start with task_id=. Tests: 2 new goal-path stage cases + a done detail task-id assertion (8 total in test_prestart_stages).
This commit is contained in:
1 parent
a17f0c7412
commit
6f0eec71b6
3 files changed
+76
-28
No files matched your search
+39
-22
@@ -1113,6 +1113,37 @@ class DockerExecutor(Executor):
|
|||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
"docker runner 不支持 env,无法按名注入任务密钥")
|
"docker runner 不支持 env,无法按名注入任务密钥")
|
||||||
|
|
||||||
|
def _docker_call_staged(self, task, cmd, timeout=None, env=None):
|
||||||
|
"""带分段日志的 docker run 调用(2026-10-29 任务 A 补漏)。
|
||||||
|
|
||||||
|
goal/goal-resume/goal-resume-breakpoint 不走 _run_in_container
|
||||||
|
(直接调 _call_runner 保完整 stdout 结论),上一批只给
|
||||||
|
_run_in_container 加的标记漏了这三条真正在用的路径——
|
||||||
|
生产实测任务就是 goal 路径,日志里一条 docker_run_ 都没有。
|
||||||
|
本 helper 统一四条任务路径的开始/结束/异常行;异常只记
|
||||||
|
类名后原样重抛,各调用点的 kill+failed 处理不变。只记
|
||||||
|
容器名与镜像,绝不记命令(内含 prompt)。
|
||||||
|
"""
|
||||||
|
_t0 = time.time()
|
||||||
|
_DOCKER_LOGGER.info(
|
||||||
|
"docker_run_start container=%s image=%s",
|
||||||
|
self.container_name(task), self.image)
|
||||||
|
try:
|
||||||
|
result = self._call_runner(cmd, timeout=timeout, env=env)
|
||||||
|
except Exception as exc:
|
||||||
|
_DOCKER_LOGGER.warning(
|
||||||
|
"docker_run_error container=%s elapsed_ms=%d error=%s",
|
||||||
|
self.container_name(task),
|
||||||
|
int((time.time() - _t0) * 1000),
|
||||||
|
exc.__class__.__name__)
|
||||||
|
raise
|
||||||
|
_DOCKER_LOGGER.info(
|
||||||
|
"docker_run_done container=%s elapsed_ms=%d returncode=%s",
|
||||||
|
self.container_name(task),
|
||||||
|
int((time.time() - _t0) * 1000),
|
||||||
|
getattr(result, "returncode", None))
|
||||||
|
return result
|
||||||
|
|
||||||
def list_containers(self, prefix="codex-"):
|
def list_containers(self, prefix="codex-"):
|
||||||
"""列出名字以 prefix 开头的容器(默认 codex-),失败返回 []。
|
"""列出名字以 prefix 开头的容器(默认 codex-),失败返回 []。
|
||||||
|
|
||||||
@@ -1198,20 +1229,11 @@ class DockerExecutor(Executor):
|
|||||||
extra_mounts=extra_mounts,
|
extra_mounts=extra_mounts,
|
||||||
network=network,
|
network=network,
|
||||||
extra_env=extra_env)
|
extra_env=extra_env)
|
||||||
_docker_t0 = time.time()
|
|
||||||
_DOCKER_LOGGER.info(
|
|
||||||
"docker_run_start container=%s image=%s",
|
|
||||||
self.container_name(task), self.image)
|
|
||||||
try:
|
try:
|
||||||
result = self._call_runner(
|
result = self._docker_call_staged(
|
||||||
cmd, timeout=timeout,
|
task, cmd, timeout=timeout,
|
||||||
env=self._docker_cli_env(extra_env))
|
env=self._docker_cli_env(extra_env))
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
_DOCKER_LOGGER.warning(
|
|
||||||
"docker_run_error container=%s elapsed_ms=%d error=%s",
|
|
||||||
self.container_name(task),
|
|
||||||
int((time.time() - _docker_t0) * 1000),
|
|
||||||
exc.__class__.__name__)
|
|
||||||
if exc.__class__.__name__ == "TimeoutExpired":
|
if exc.__class__.__name__ == "TimeoutExpired":
|
||||||
self.kill_container(task)
|
self.kill_container(task)
|
||||||
return {
|
return {
|
||||||
@@ -1231,11 +1253,6 @@ class DockerExecutor(Executor):
|
|||||||
parsed = _parse_host_result(task_id, fake)
|
parsed = _parse_host_result(task_id, fake)
|
||||||
parsed["executor"] = "docker"
|
parsed["executor"] = "docker"
|
||||||
parsed["container"] = self.container_name(task)
|
parsed["container"] = self.container_name(task)
|
||||||
_DOCKER_LOGGER.info(
|
|
||||||
"docker_run_done container=%s elapsed_ms=%d returncode=%s",
|
|
||||||
self.container_name(task),
|
|
||||||
int((time.time() - _docker_t0) * 1000),
|
|
||||||
parsed.get("returncode"))
|
|
||||||
return parsed
|
return parsed
|
||||||
|
|
||||||
def dispatch_task(self, task, timeout=None, extra_mounts=None,
|
def dispatch_task(self, task, timeout=None, extra_mounts=None,
|
||||||
@@ -1357,8 +1374,8 @@ class DockerExecutor(Executor):
|
|||||||
network=network,
|
network=network,
|
||||||
extra_env=extra_env)
|
extra_env=extra_env)
|
||||||
try:
|
try:
|
||||||
result = self._call_runner(
|
result = self._docker_call_staged(
|
||||||
cmd, timeout=timeout,
|
task, cmd, timeout=timeout,
|
||||||
env=self._docker_cli_env(extra_env))
|
env=self._docker_cli_env(extra_env))
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
if exc.__class__.__name__ == "TimeoutExpired":
|
if exc.__class__.__name__ == "TimeoutExpired":
|
||||||
@@ -1472,8 +1489,8 @@ class DockerExecutor(Executor):
|
|||||||
network=network,
|
network=network,
|
||||||
extra_env=extra_env)
|
extra_env=extra_env)
|
||||||
try:
|
try:
|
||||||
result = self._call_runner(
|
result = self._docker_call_staged(
|
||||||
cmd, timeout=timeout,
|
task, cmd, timeout=timeout,
|
||||||
env=self._docker_cli_env(extra_env))
|
env=self._docker_cli_env(extra_env))
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
if exc.__class__.__name__ == "TimeoutExpired":
|
if exc.__class__.__name__ == "TimeoutExpired":
|
||||||
@@ -1582,8 +1599,8 @@ class DockerExecutor(Executor):
|
|||||||
network=network,
|
network=network,
|
||||||
extra_env=extra_env)
|
extra_env=extra_env)
|
||||||
try:
|
try:
|
||||||
result = self._call_runner(
|
result = self._docker_call_staged(
|
||||||
cmd, timeout=timeout,
|
task, cmd, timeout=timeout,
|
||||||
env=self._docker_cli_env(extra_env))
|
env=self._docker_cli_env(extra_env))
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
if exc.__class__.__name__ == "TimeoutExpired":
|
if exc.__class__.__name__ == "TimeoutExpired":
|
||||||
|
|||||||
@@ -85,6 +85,15 @@ class PrestartStagesTestCase(unittest.TestCase):
|
|||||||
self.assertIn("worker_exec_start", actions)
|
self.assertIn("worker_exec_start", actions)
|
||||||
self.assertIn("worker_exec_done", actions)
|
self.assertIn("worker_exec_done", actions)
|
||||||
self.assertNotIn("worker_exec_error", actions)
|
self.assertNotIn("worker_exec_error", actions)
|
||||||
|
# 审计表没有任务 id 列,detail 必须自带任务 id 才能对应。
|
||||||
|
conn = sqlite3.connect(self.db_path)
|
||||||
|
try:
|
||||||
|
detail = conn.execute(
|
||||||
|
"SELECT detail FROM audit_log"
|
||||||
|
" WHERE action = 'worker_exec_done'").fetchone()[0]
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
self.assertIn("T1", detail)
|
||||||
|
|
||||||
def test_exec_error_audited_not_done(self):
|
def test_exec_error_audited_not_done(self):
|
||||||
self.worker.executor = FakeExecutor(
|
self.worker.executor = FakeExecutor(
|
||||||
@@ -159,6 +168,28 @@ class DockerRunStagesTestCase(unittest.TestCase):
|
|||||||
self.assertTrue(any("docker_run_error" in r
|
self.assertTrue(any("docker_run_error" in r
|
||||||
for r in logs.output))
|
for r in logs.output))
|
||||||
|
|
||||||
|
def test_goal_task_stage_lines_logged(self):
|
||||||
|
# goal 路径不走 _run_in_container(直接调 runner 保完整
|
||||||
|
# stdout 结论),上一批的标记漏了这条真正在用的路径。
|
||||||
|
def runner(cmd, timeout=None, env=None):
|
||||||
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||||
|
ex = self._executor(runner)
|
||||||
|
with self.assertLogs("cloud.docker", level="INFO") as logs:
|
||||||
|
ex.goal_task(self._task())
|
||||||
|
joined = "\n".join(logs.output)
|
||||||
|
self.assertIn("docker_run_start", joined)
|
||||||
|
self.assertIn("docker_run_done", joined)
|
||||||
|
|
||||||
|
def test_goal_resume_breakpoint_stage_lines_logged(self):
|
||||||
|
def runner(cmd, timeout=None, env=None):
|
||||||
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||||
|
ex = self._executor(runner)
|
||||||
|
with self.assertLogs("cloud.docker", level="INFO") as logs:
|
||||||
|
ex.goal_resume_breakpoint_task(self._task())
|
||||||
|
joined = "\n".join(logs.output)
|
||||||
|
self.assertIn("docker_run_start", joined)
|
||||||
|
self.assertIn("docker_run_done", joined)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
+6
-6
@@ -3560,8 +3560,8 @@ class Worker:
|
|||||||
_exec_container = None
|
_exec_container = None
|
||||||
_exec_t0 = time.time()
|
_exec_t0 = time.time()
|
||||||
_exec_error = None
|
_exec_error = None
|
||||||
_exec_detail = "kind=%s executor=%s timeout=%s" % (
|
_exec_detail = "task_id=%s kind=%s executor=%s timeout=%s" % (
|
||||||
_exec_kind, self._executor_name(),
|
task_id, _exec_kind, self._executor_name(),
|
||||||
"None" if _effective_task_timeout is None
|
"None" if _effective_task_timeout is None
|
||||||
else _effective_task_timeout)
|
else _effective_task_timeout)
|
||||||
if _exec_container:
|
if _exec_container:
|
||||||
@@ -3797,8 +3797,8 @@ class Worker:
|
|||||||
_exec_elapsed_ms = int((time.time() - _exec_t0) * 1000)
|
_exec_elapsed_ms = int((time.time() - _exec_t0) * 1000)
|
||||||
if _exec_error is not None:
|
if _exec_error is not None:
|
||||||
self._audit(task.get("project_id"), "worker_exec_error",
|
self._audit(task.get("project_id"), "worker_exec_error",
|
||||||
"elapsed_ms=%d error=%s"
|
"task_id=%s elapsed_ms=%d error=%s"
|
||||||
% (_exec_elapsed_ms,
|
% (task_id, _exec_elapsed_ms,
|
||||||
_exec_error.__class__.__name__))
|
_exec_error.__class__.__name__))
|
||||||
log_event(self.logger, "worker_exec_error", task_id=task_id,
|
log_event(self.logger, "worker_exec_error", task_id=task_id,
|
||||||
message="worker exec error: elapsed_ms=%d error=%s"
|
message="worker exec error: elapsed_ms=%d error=%s"
|
||||||
@@ -3806,8 +3806,8 @@ class Worker:
|
|||||||
_exec_error.__class__.__name__))
|
_exec_error.__class__.__name__))
|
||||||
else:
|
else:
|
||||||
self._audit(task.get("project_id"), "worker_exec_done",
|
self._audit(task.get("project_id"), "worker_exec_done",
|
||||||
"elapsed_ms=%d status=%s returncode=%s"
|
"task_id=%s elapsed_ms=%d status=%s returncode=%s"
|
||||||
% (_exec_elapsed_ms, result.get("status"),
|
% (task_id, _exec_elapsed_ms, result.get("status"),
|
||||||
result.get("returncode")))
|
result.get("returncode")))
|
||||||
log_event(self.logger, "worker_exec_done", task_id=task_id,
|
log_event(self.logger, "worker_exec_done", task_id=task_id,
|
||||||
status=result.get("status"),
|
status=result.get("status"),
|
||||||
|
|||||||
Reference in new issue
Block a user