Files
infrapuzzle/k8s/openclaw/scripts/sync_crons.py
T

138 lines
4.5 KiB
Python

#!/usr/bin/env python3
"""
sync_crons.py
Idempotently reconciles declarative OpenClaw automations defined in crons.yaml
with the running OpenClaw gateway state.
Can run directly inside the container or remotely via kubectl exec.
"""
import json
import os
import subprocess
import sys
import yaml
def find_crons_file():
candidates = [
os.environ.get("CRONS_FILE"),
"crons.yaml",
"k8s/openclaw/crons.yaml",
"/home/node/.openclaw/scripts/crons.yaml",
os.path.join(os.path.dirname(__file__), "../crons.yaml"),
os.path.join(os.path.dirname(__file__), "crons.yaml"),
]
for c in candidates:
if c and os.path.exists(c):
return os.path.abspath(c)
raise FileNotFoundError("Could not find crons.yaml. Pass via CRONS_FILE environment variable.")
def is_inside_container():
return os.path.exists("/home/node/.openclaw") or os.environ.get("OPENCLAW_HOME") is not None
def run_openclaw_cmd(args):
if is_inside_container():
cmd = ["openclaw"] + args
else:
namespace = os.environ.get("NAMESPACE", "openclaw")
pod = os.environ.get("POD_NAME", "openclaw-0")
container = os.environ.get("CONTAINER_NAME", "openclaw")
cmd = ["kubectl", "exec", "-n", namespace, pod, "-c", container, "--", "openclaw"] + args
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode != 0:
raise RuntimeError(f"OpenClaw CLI command failed ({' '.join(cmd)}):\n{res.stderr}\n{res.stdout}")
return res.stdout
def list_existing_jobs():
raw = run_openclaw_cmd(["cron", "list", "--json"])
try:
data = json.loads(raw)
return data.get("jobs", [])
except Exception as e:
print(f"Warning: could not parse cron list json: {e}\nRaw output: {raw[:300]}", file=sys.stderr)
return []
def reconcile_jobs():
file_path = find_crons_file()
print(f"=== Syncing OpenClaw Crons-as-Code from {file_path} ===")
with open(file_path, "r", encoding="utf-8") as f:
spec = yaml.safe_load(f)
declarations = spec.get("jobs", [])
existing_jobs = list_existing_jobs()
print(f"Found {len(declarations)} declared jobs in YAML, {len(existing_jobs)} active jobs on gateway.")
for job in declarations:
name = job.get("name")
decl_key = job.get("declarationKey") or name
desc = job.get("description", "")
sched = job.get("schedule", {})
# Match existing job by declarationKey or exact name
match = next((j for j in existing_jobs if j.get("declarationKey") == decl_key or j.get("name") == name), None)
if match:
existing_id = match.get("id")
print(f"Removing existing job '{name}' (ID {existing_id}) for idempotent recreation...")
try:
run_openclaw_cmd(["cron", "rm", existing_id])
except Exception as e:
print(f"Warning removing {existing_id}: {e}")
# Build add command
cmd = [
"cron", "add",
"--name", name,
"--declaration-key", decl_key,
"--description", desc,
]
if sched.get("kind") == "every":
cmd.extend(["--every", sched.get("every", "15m")])
elif sched.get("kind") == "cron":
cmd.extend(["--cron", sched.get("expr", "* * * * *")])
if sched.get("tz"):
cmd.extend(["--tz", sched.get("tz")])
if job.get("session"):
cmd.extend(["--session", job.get("session")])
if job.get("delivery") == "none":
cmd.append("--no-deliver")
if job.get("command"):
cmd.extend(["--command", job.get("command")])
if job.get("agent"):
cmd.extend(["--agent", job.get("agent")])
if job.get("model"):
cmd.extend(["--model", job.get("model")])
if job.get("channel"):
cmd.extend(["--channel", job.get("channel")])
if job.get("to"):
cmd.extend(["--to", job.get("to")])
if job.get("threadId"):
cmd.extend(["--thread-id", str(job.get("threadId"))])
if job.get("announce"):
cmd.append("--announce")
if job.get("message"):
cmd.extend(["--message", job.get("message")])
print(f"Registering job '{name}'...")
run_openclaw_cmd(cmd)
print(f"Successfully registered '{name}'.")
print("=== OpenClaw Crons-as-Code reconciliation complete! ===")
if __name__ == "__main__":
reconcile_jobs()