Tom Han
10/01/2025, 8:27 PMPREFECT_RUNNER_HEARTBEAT_FREQUENCY something I set in my server or for each runner?Marvin
10/01/2025, 8:27 PMMarvin
10/01/2025, 8:29 PMPREFECT_RUNNER_HEARTBEAT_FREQUENCY is a client-side setting read by the flow runner process. Set it in the environment where each runner executes (e.g., your Docker container, K8s pod, VM), or via the Prefect profile active for that process. The server/Cloud does not control this value.
Quick ways to set it:
- Shell/env: export PREFECT_RUNNER_HEARTBEAT_FREQUENCY=30
- Prefect profile (on the machine that runs flows): add it to your active profile in ~/.prefect/profiles.toml
- In container/orchestrator: add it as an environment variable in your deployment spec
Notes:
- Default is 30 seconds; missed heartbeats help detect “zombie” flow runs.
- Changing the value requires restarting the runner process to take effect.
Docs: Detect zombie flows (heartbeats and automations)Tom Han
10/02/2025, 1:57 AMfrom datetime import timedelta
...:
...: from prefect.automations import Automation
...: from prefect.client.schemas.objects import StateType
...: from prefect.events.actions import ChangeFlowRunState
...: from prefect.events.schemas.automations import EventTrigger, Posture
...: from prefect.events.schemas.events import ResourceSpecification
...:
...:
...: my_automation = Automation(
...: name="Crash zombie flows",
...: trigger=EventTrigger(
...: after={"prefect.flow-run.heartbeat"},
...: expect={
...: "prefect.flow-run.heartbeat",
...: "prefect.flow-run.Completed",
...: "prefect.flow-run.Failed",
...: "prefect.flow-run.Cancelled",
...: "prefect.flow-run.Crashed",
...: },
...: match=ResourceSpecification({"prefect.resource.id": ["prefect.flow-run.*"]}),
...: for_each={"prefect.resource.id"},
...: posture=Posture.Proactive,
...: threshold=1,
...: within=timedelta(seconds=90),
...: ),
...: actions=[
...: ChangeFlowRunState(
...: state=StateType.CRASHED,
...: message="Flow run marked as crashed due to missing heartbeats.",
...: )
...: ],
...: )
...:
...: if __name__ == "__main__":
...: my_automation.create()
But prefect is still not pruning zombie flows. I didn't set the PREFECT_RUNNER_HEARTBEAT_FREQ because I thought it is a default value. Am I wrong?Marvin
10/02/2025, 1:57 AMMarvin
10/02/2025, 2:00 AMPREFECT_RUNNER_HEARTBEAT_FREQUENCY (not ..._FREQ).
- In Prefect 3.x, the default is effectively “off” (it’s None). You must set it to a value >= 30 seconds to enable heartbeats.
- It’s a per-runner setting, so set it in the environment where your flows actually run, then restart those processes.
What to do:
1) Set the env var where your flows run and restart:
export PREFECT_RUNNER_HEARTBEAT_FREQUENCY=30
2) Verify heartbeats are being emitted by checking for “prefect.flow-run.heartbeat” events in the UI/events log. You can also sanity check from code:
from prefect.settings import get_current_settings
print(get_current_settings().runner.heartbeat_frequency)
A small improvement to your automation:
- If a flow crashes before the first heartbeat (e.g., crashes in <30s), your after={"prefect.flow-run.heartbeat"} won’t start the timer. Consider starting the timer on running instead:
after={"prefect.flow-run.Running"},
expect={
"prefect.flow-run.heartbeat",
"prefect.flow-run.Completed",
"prefect.flow-run.Failed",
"prefect.flow-run.Cancelled",
"prefect.flow-run.Crashed",
},
within=timedelta(seconds=90),
threshold=1,
This catches flows that never send a first heartbeat.
Docs: Detect zombie flows (heartbeats and automations)Tom Han
10/02/2025, 2:02 AMMarvin
10/02/2025, 2:02 AMMarvin
10/02/2025, 2:04 AMhandle_engine_signals) so when it receives SIGTERM/SIGINT it will update the flow run (typically to CRASHED) before exiting.
- If you send the signal to some parent that doesn’t propagate it, the flow process won’t see it and the run can become a zombie until heartbeats/automations catch it.
Practical guidance by environment:
- Docker/containers
- Make sure the Python flow process is PID 1 or signals are forwarded to it.
- Use exec-form ENTRYPOINT/CMD so Python replaces the shell:
CMD ["python", "-m", "your_module"]
- Or enable init so signals are forwarded: docker run --init … or init: true in Compose.
- Prefer SIGTERM and give a short grace period so Prefect can write the final state.
- Kubernetes
- K8s sends SIGTERM to PID 1. Ensure your container runs the Python flow process as PID 1 (exec-form) or uses an init like tini that forwards signals.
- Set terminationGracePeriodSeconds (e.g., 30–60s) to allow state update before SIGKILL.
- Systemd or bare metal
- Send SIGTERM/SIGINT directly to the flow process PID (the one logging “Starting flow run …”). Avoid killing only a supervisor that doesn’t forward signals.
Notes:
- SIGKILL (9) cannot be handled; Prefect can’t set state on SIGKILL. In those cases, rely on heartbeats and your automation to mark runs as crashed.
- If you’re using heartbeats for zombie detection, keep PREFECT_RUNNER_HEARTBEAT_FREQUENCY set (>=30) so missed-heartbeat automations backstop any unhandled terminations: Detect zombie flows (heartbeats and automations)
If you can share how you’re running flows (Docker, K8s, a runner server, etc.), I can give concrete setup snippets to ensure signals reach the flow process.Tom Han
10/02/2025, 4:58 AMMarvin
10/02/2025, 4:58 AMMarvin
10/02/2025, 5:00 AMPREFECT_RUNNER_HEARTBEAT_FREQUENCY after your ProcessWorker (or runner) is already running, it won’t take effect for that process or any already-running flow runs.
- Restart the worker/runner so it picks up the env var, then new flow runs will emit heartbeats.
- Existing runs that started without heartbeats won’t “switch on” mid-run.
Tips:
- Verify it’s active inside the running process:
from prefect.settings import get_current_settings
print(get_current_settings().runner.heartbeat_frequency) # expect >= 30
- In containers/K8s, add the env var to the pod/container spec and roll out a restart.
- Make sure the name is exact: PREFECT_RUNNER_HEARTBEAT_FREQUENCY.
Docs: Detect zombie flows (heartbeats and automations)