execution_log.py
8.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
"""c_execution_log_get + c_ws_log_tail (03-C §3.6.1/§3.6.2) + p_log_retention_cleanup
(04-P §4.5): raw execution log storage, retrieval, real-time streaming, and
scheduled cleanup -- REQ-13.
U-013 (work_node_run.py) already manages execution_log_file's row lifecycle
(insert status='active' -> update status='closed') but nothing ever wrote
actual log content -- append_log() is that missing write path, meant to be
called from inside a run_fn as it works (this module doesn't know or care
what a node's actual work is, same boundary work_node_run.py itself draws
around run_fn).
p_log_retention_cleanup is a scheduled job (04-P: "定时触发...具体周期由
平台调度配置，非用户可配") -- no scheduler daemon exists in this codebase
yet and building one is out of this Unit's scope (coverage only lists
c_execution_log_get/c_ws_log_tail); run_log_retention_cleanup() is the pure,
directly-testable function a future scheduler calls periodically.
"""
import json
import os
import uuid
from datetime import datetime, timedelta, timezone
class LogNotFoundError(Exception):
"""LOG_NOT_FOUND -- 03-C §3.6.1: node_exec_id has no execution_log_file row."""
class LogPurgedError(Exception):
"""LOG_PURGED -- 03-C §3.6.1: the log was already cleaned up by retention."""
class NodeExecNotFoundError(Exception):
"""NODE_EXEC_NOT_FOUND -- 03-C §3.6.2: c_ws_log_tail subscribe target doesn't exist."""
_TERMINAL_NODE_STATUSES = {"done", "blocked", "failed"}
def is_terminal_status(status):
"""02-V §2.7.1 / 03-C §3.6.2: once a node reaches one of these, no more
log content will ever be appended -- c_ws_log_tail actively closes."""
return status in _TERMINAL_NODE_STATUSES
def append_log(conn, node_exec_id, content):
"""Appends `content` to the node's log file on disk, bumps
execution_log_file.size_bytes, and writes
execution_event(event_type='log_appended') so c_ws_log_tail subscribers
see it in realtime mode (02-V §2.7.1)."""
row = conn.execute(
"SELECT log_file_id, project_id, file_path FROM execution_log_file WHERE node_exec_id = ?",
(node_exec_id,),
).fetchone()
if row is None:
raise LogNotFoundError(node_exec_id)
log_file_id, project_id, file_path = row
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "a", encoding="utf-8") as f:
f.write(content)
new_size = os.path.getsize(file_path)
conn.execute("UPDATE execution_log_file SET size_bytes = ? WHERE log_file_id = ?", (new_size, log_file_id))
task_id = conn.execute(
"SELECT wgr.task_id FROM work_node_execution wne "
"JOIN work_graph_run wgr ON wgr.run_id = wne.run_id WHERE wne.node_exec_id = ?",
(node_exec_id,),
).fetchone()[0]
conn.execute(
"INSERT INTO execution_event (event_id, project_id, task_id, node_exec_id, event_type, payload) "
"VALUES (?, ?, ?, ?, 'log_appended', ?)",
(str(uuid.uuid4()), project_id, task_id, node_exec_id, json.dumps({"appended_content": content})),
)
conn.commit()
def get_log(conn, node_exec_id, offset=0, limit=None):
"""c_execution_log_get (03-C §3.6.1) static-mode read, paginated by
byte offset/limit so a large file isn't loaded all at once."""
row = conn.execute(
"SELECT file_path, format, status FROM execution_log_file WHERE node_exec_id = ?", (node_exec_id,)
).fetchone()
if row is None:
raise LogNotFoundError(node_exec_id)
file_path, fmt, status = row
if status == "purged":
raise LogPurgedError(node_exec_id)
if not os.path.isfile(file_path):
return {"content": "", "format": fmt, "total_size_bytes": 0, "status": status}
total_size = os.path.getsize(file_path)
with open(file_path, "rb") as f:
f.seek(offset)
chunk = f.read(limit) if limit is not None else f.read()
return {"content": chunk.decode("utf-8", errors="replace"), "format": fmt, "total_size_bytes": total_size, "status": status}
def get_log_file_path(conn, node_exec_id):
"""For the download=true response path (send the raw file, not JSON)."""
row = conn.execute(
"SELECT file_path, status FROM execution_log_file WHERE node_exec_id = ?", (node_exec_id,)
).fetchone()
if row is None:
raise LogNotFoundError(node_exec_id)
file_path, status = row
if status == "purged":
raise LogPurgedError(node_exec_id)
return file_path
def node_exec_status(conn, node_exec_id):
row = conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,)).fetchone()
if row is None:
raise NodeExecNotFoundError(node_exec_id)
return row[0]
def latest_log_event_rowid(conn, node_exec_id):
row = conn.execute(
"SELECT MAX(rowid) FROM execution_event WHERE node_exec_id = ? AND event_type = 'log_appended'",
(node_exec_id,),
).fetchone()
return row[0] or 0
def poll_log_events(conn, node_exec_id, after_rowid):
"""c_ws_log_tail's push source -- same rowid-increment polling pattern
as ws_subscribe.poll_new_events, scoped to one node_exec_id's
log_appended events."""
rows = conn.execute(
"SELECT rowid, payload, created_at FROM execution_event "
"WHERE node_exec_id = ? AND event_type = 'log_appended' AND rowid > ? ORDER BY rowid",
(node_exec_id, after_rowid),
).fetchall()
events = [
{
"node_exec_id": node_exec_id,
"appended_content": json.loads(r[1])["appended_content"],
"created_at": r[2],
}
for r in rows
]
new_after_rowid = rows[-1][0] if rows else after_rowid
return events, new_after_rowid
def run_log_retention_cleanup(conn, now=None, config_id="default"):
"""p_log_retention_cleanup (04-P §4.5): idle -> scanning -> purging -> idle.
Age-expired 'closed' logs are always candidates. If max_total_size_mb is
set and the remaining (non-expired) total still exceeds it, the
oldest-first remaining logs are added as candidates too, until the kept
total drops back under the cap. A file that fails to delete (e.g. still
open elsewhere) is recorded as a failure and left status='closed' for
the next cycle to retry -- it must not block purging the rest of the
batch (REQ-13 failure case).
Returns {"purged": [log_file_id, ...], "failed": [log_file_id, ...]}.
"""
config_row = conn.execute(
"SELECT retention_days, max_total_size_mb FROM log_retention_config WHERE config_id = ?", (config_id,)
).fetchone()
if config_row is None:
raise ValueError("no such log_retention_config: " + config_id)
retention_days, max_total_size_mb = config_row
now = now if now is not None else datetime.now(timezone.utc)
cutoff_str = (now - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S")
expired = conn.execute(
"SELECT log_file_id, file_path, size_bytes FROM execution_log_file "
"WHERE status = 'closed' AND created_at < ? ORDER BY created_at",
(cutoff_str,),
).fetchall()
kept = conn.execute(
"SELECT log_file_id, file_path, size_bytes FROM execution_log_file "
"WHERE status = 'closed' AND created_at >= ? ORDER BY created_at",
(cutoff_str,),
).fetchall()
candidates = list(expired)
if max_total_size_mb is not None:
limit_bytes = max_total_size_mb * 1024 * 1024
kept_total = sum(row[2] for row in kept)
for row in kept:
if kept_total <= limit_bytes:
break
candidates.append(row)
kept_total -= row[2]
purged, failed = [], []
for log_file_id, file_path, _size in candidates:
try:
if os.path.isfile(file_path):
os.remove(file_path)
elif os.path.exists(file_path):
# something's occupying the path but isn't a plain file
# (e.g. a directory) -- can't silently treat that as "already
# gone", it's a real delete failure
raise OSError("not a regular file: " + file_path)
except OSError:
failed.append(log_file_id)
continue
conn.execute(
"UPDATE execution_log_file SET status = 'purged', purged_at = CURRENT_TIMESTAMP WHERE log_file_id = ?",
(log_file_id,),
)
purged.append(log_file_id)
conn.commit()
return {"purged": purged, "failed": failed}