60 lines
1.9 KiB
Python
Executable File
60 lines
1.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Requeue Cassandra workers that were interrupted by a Hermes pod restart."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
|
|
BOARD = "cassandra"
|
|
REASON = "Hermes pod restarted; resume the durable task from its saved context"
|
|
|
|
|
|
def recover_running_tasks(kanban_db: Any) -> list[str]:
|
|
"""Reclaim every running Cassandra task and return the recovered task IDs."""
|
|
if not kanban_db.board_exists(BOARD):
|
|
return []
|
|
|
|
recovered: list[str] = []
|
|
with kanban_db.scoped_current_board(BOARD):
|
|
connection = kanban_db.connect(board=BOARD)
|
|
try:
|
|
for task in kanban_db.list_tasks(connection):
|
|
if str(getattr(task, "status", "")) != "running":
|
|
continue
|
|
task_id = str(getattr(task, "id", ""))
|
|
if not task_id or not kanban_db.reclaim_task(
|
|
connection,
|
|
task_id,
|
|
reason=REASON,
|
|
):
|
|
continue
|
|
kanban_db.add_comment(
|
|
connection,
|
|
task_id,
|
|
"pod-recovery",
|
|
"Requeued after the Hermes pod restart; the replacement pod will resume this task.",
|
|
)
|
|
recovered.append(task_id)
|
|
finally:
|
|
connection.close()
|
|
return recovered
|
|
|
|
|
|
def main() -> int:
|
|
"""Load Hermes's Kanban API and reclaim interrupted Cassandra work."""
|
|
os.environ.setdefault("HERMES_HOME", "/opt/data")
|
|
from hermes_cli import kanban_db
|
|
|
|
recovered = recover_running_tasks(kanban_db)
|
|
if recovered:
|
|
print(f"Requeued {len(recovered)} interrupted Cassandra task(s): {', '.join(recovered)}")
|
|
else:
|
|
print("No interrupted Cassandra tasks needed recovery")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|