Add weekly coding retro dry-run scheduler script
Wire ACTIVITY-WP-0008 dry-run helper and update retro activity definition, runbook, and workplan notes.
This commit is contained in:
parent
f314bfc7f2
commit
f171bb037a
4 changed files with 129 additions and 6 deletions
112
scripts/schedule_coding_retro_dry_run.py
Normal file
112
scripts/schedule_coding_retro_dry_run.py
Normal file
|
|
@ -0,0 +1,112 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Schedule delayed manual triggers for weekly-coding-retro dry-run verification."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--activity-id", required=True)
|
||||
parser.add_argument("--api-base", default=os.environ.get(
|
||||
"ACTCORE_API_URL",
|
||||
"http://actcore-api.activity-core.svc.cluster.local:8010",
|
||||
))
|
||||
parser.add_argument("--temporal-host", default=os.environ.get("TEMPORAL_HOST", "localhost:7233"))
|
||||
parser.add_argument("--temporal-namespace", default=os.environ.get("TEMPORAL_NAMESPACE", "default"))
|
||||
parser.add_argument("--delay-seconds", type=int, default=600)
|
||||
parser.add_argument("--timeout-seconds", type=int, default=600)
|
||||
parser.add_argument("--runs", type=int, default=2, help="Number of back-to-back trigger runs")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def api_json(api_base: str, path: str, *, method: str = "GET", body: Any | None = None) -> Any:
|
||||
payload = json.dumps(body).encode() if body is not None else None
|
||||
req = urllib.request.Request(
|
||||
f"{api_base.rstrip('/')}{path}",
|
||||
data=payload,
|
||||
headers={"Content-Type": "application/json"} if payload else {},
|
||||
method=method,
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
|
||||
|
||||
async def wait_workflow(client: Any, workflow_id: str, timeout_seconds: int) -> dict[str, Any]:
|
||||
deadline = time.time() + timeout_seconds
|
||||
last_error: str | None = None
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
handle = client.get_workflow_handle(workflow_id)
|
||||
desc = await handle.describe()
|
||||
status = str(desc.status)
|
||||
if status == "2":
|
||||
return {"status": "completed", "result": await handle.result()}
|
||||
if status != "1":
|
||||
return {"status": status, "error": f"workflow ended with status {status}"}
|
||||
except Exception as exc: # noqa: BLE001 - operator script
|
||||
last_error = str(exc)
|
||||
await asyncio.sleep(3)
|
||||
return {"status": "timeout", "error": last_error or "workflow did not complete"}
|
||||
|
||||
|
||||
async def run(args: argparse.Namespace) -> dict[str, Any]:
|
||||
fire_at = datetime.now(timezone.utc).timestamp() + args.delay_seconds
|
||||
report: dict[str, Any] = {
|
||||
"activity_id": args.activity_id,
|
||||
"delay_seconds": args.delay_seconds,
|
||||
"fire_at_unix": fire_at,
|
||||
"runs": [],
|
||||
}
|
||||
print(json.dumps({"phase": "scheduled", **report}), flush=True)
|
||||
await asyncio.sleep(args.delay_seconds)
|
||||
|
||||
from temporalio.client import Client
|
||||
|
||||
client = await Client.connect(args.temporal_host, namespace=args.temporal_namespace)
|
||||
for index in range(args.runs):
|
||||
label = f"dry_run_{index + 1}"
|
||||
trigger = api_json(args.api_base, f"/activity-definitions/{args.activity_id}/trigger", method="POST")
|
||||
workflow_id = str(trigger["workflow_id"])
|
||||
print(json.dumps({"phase": label, "trigger": trigger}), flush=True)
|
||||
wait = await wait_workflow(client, workflow_id, args.timeout_seconds)
|
||||
entry = {"label": label, "workflow_id": workflow_id, "wait": wait}
|
||||
report["runs"].append(entry)
|
||||
print(json.dumps({"phase": label, "wait": wait}), flush=True)
|
||||
if wait.get("status") != "completed":
|
||||
report["status"] = "failed"
|
||||
return report
|
||||
if index + 1 < args.runs:
|
||||
await asyncio.sleep(5)
|
||||
|
||||
report["status"] = "completed"
|
||||
return report
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
try:
|
||||
report = asyncio.run(run(args))
|
||||
except urllib.error.HTTPError as exc:
|
||||
body = exc.read().decode(errors="replace")
|
||||
print(json.dumps({"status": "failed", "error": str(exc), "body": body}), file=sys.stderr)
|
||||
return 2
|
||||
except Exception as exc: # noqa: BLE001 - operator script
|
||||
print(json.dumps({"status": "failed", "error": str(exc)}), file=sys.stderr)
|
||||
return 2
|
||||
print(json.dumps({"phase": "done", **report}), flush=True)
|
||||
return 0 if report.get("status") == "completed" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Add table
Add a link
Reference in a new issue