Thanks for laying this out so clearly — you’ve hit a few important edge cases. Here’s what’s going on in Prefect 3 with Kubernetes workers, and how to make cancellation and hard time limits reliable. High-level answers to your points - Your assessment of

timeout_seconds

is correct. In Prefect 3, timeouts and cancellations are cooperative at the Python level. They will not interrupt hung C extensions, stuck CUDA/GPU calls, or blockingly-waiting subprocesses; they only take effect when Python regains control. - Kubernetes’

activeDeadlineSeconds

is a hard, kubelet-enforced kill and is not controlled by Prefect. When it fires, the pod is terminated and the Job is marked failed. Prefect does not recreate the pod itself — any “requeue” you’re seeing is almost always due to flow-level retries, an automation that runs on Crash, or an observer/worker race/bug. - The Prefect Kubernetes worker does include a Kopf-based observer to watch Jobs/Pods and propose final states (e.g., Crashed). If the observer crashes while handling a failure, you can see “ghost”-like requeues or state flapping if you also have retries/automations in play. - Dashboard “Cancel (cascade)” ultimately calls the worker’s kill routine for the Job, which tells Kubernetes to delete the Job with a grace period and a foreground propagation policy. If the process inside the container is uninterruptible, kubelet will escalate to SIGKILL after

terminationGracePeriodSeconds

. If the underlying process is truly unkillable (D-state/kernel/GPU driver hang), even Kubernetes cannot end it — Prefect can’t paper over that. What typically causes “ghost requeues” - Flow-level retries > 0 on the flow or tasks - A Prefect Automation that fires “On Flow Run Crashed → Run Flow” - Kubernetes Job

backoffLimit > 0

(Kubernetes itself will retry the pod) - An observer/worker race where the observer restarts while proposing crash, and a retry/automation resubmits the run How to enforce hard limits and reliable cancellation with the k8s worker 1) Use Kubernetes for hard stops (ADS + short grace) - Set a Job-level active deadline and short grace so kubelet escalates to SIGKILL quickly:

spec:
  backoffLimit: 0
  activeDeadlineSeconds: 3600   # hard limit
  template:
    spec:
      restartPolicy: Never
      terminationGracePeriodSeconds: 30

- Ensure you are not also telling Kubernetes to retry (

backoffLimit: 0

is important). 2) Ensure Prefect won’t resubmit - Set flow-level retries to 0 (and avoid task retries if you want a hard stop). - Check for any Automations that trigger on Crashed (e.g., “On Flow Run Crashed → Run Flow”) and disable them for these runs. - If you have multiple workers with observers enabled, ensure you’re not double-watching the same namespaces. 3) Make deletion deterministic when cancelling from Prefect - The Kubernetes worker’s cancel path deletes the Job with a grace period and a foreground propagation policy, which should cascade to the Pod. You can lower the grace period if you want faster escalation to SIGKILL by overriding the worker’s Job template delete behavior (or setting a lower

terminationGracePeriodSeconds

so kubelet does it for you). 4) Handle uninterruptible work - If you have known uninterruptible calls (CUDA, some C extensions, stuck subprocess): - Keep

terminationGracePeriodSeconds

short (15–30s) so kubelet escalates to SIGKILL quickly. - Consider adding a simple preStop to give logs a chance to flush before SIGKILL:

lifecycle:
      preStop:
        exec:
          command: ["/bin/sh", "-c", "sleep 5"]

- If you need a belt-and-suspenders approach, add a watchdog sidecar that can SIGKILL the process group after a fixed wall clock. 5) Observe and tune the observer - If you rely on the observer to mark runs as Crashed, make sure it’s enabled and healthy. Raise logging on the worker/observer while you diagnose: - Set

PREFECT_LOGGING_LEVEL=DEBUG

- If the observer is crashing when a Job dies due to

activeDeadlineSeconds

, please share the observer logs — that’s not expected and may be a bug we should fix in

prefect-kubernetes

. Checks and info that will help us pinpoint the “ghost requeue” - The Job/Pod template you’re using from the work pool (especially:

backoffLimit

,

activeDeadlineSeconds

,

restartPolicy

,

terminationGracePeriodSeconds

) - The flow’s retry settings and whether any Automations fire on Crash - The flow run state timeline from the UI around the time the pod is killed (e.g., Running → Crashed → Scheduled?) - Worker logs around cancellation, and any observer logs that mention the Job name (look for “marking flow run as crashed”, “job exceeded backoff limit”, etc.) - Whether you are running Ray inside the pod — Ray can keep child processes alive longer and complicate shutdown A minimal, hardened Job template snippet You can drop this into your Kubernetes worker’s job template (or pod template as appropriate) to keep kubelet in charge of hard stops and prevent retries:

apiVersion: batch/v1
kind: Job
spec:
  backoffLimit: 0
  ttlSecondsAfterFinished: 600
  activeDeadlineSeconds: 3600
  template:
    spec:
      restartPolicy: Never
      terminationGracePeriodSeconds: 30
      containers:
      - name: flow
        image: <your-image>
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            memory: "2Gi"
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 5"]

If you share the above details, I can help you confirm whether something in Prefect is resubmitting the run versus k8s retrying it, or if the observer is crashing in a specific path we need to fix. References - Prefect Kubernetes collection docs: prefect-kubernetes docs - Kubernetes Jobs termination/deadlines: Kubernetes Job concepts (see “Job termination and cleanup”) - Prefect cancellations/timeouts (background): Prefect 3 docs (see Cancellations/Timeouts concepts)

Thanks for sharing the template — this is close, but there are a few important fixes to make it behave the way you intend (especially around where certain fields belong in a Job vs. Pod spec, and a couple of gotchas that can cause unexpected retries or failures to cancel). Key adjustments I recommend - Move Job-level fields out of the Pod template: -

completions

,

parallelism

, and

podFailurePolicy

belong under

job_manifest.spec

(JobSpec), not under

template.spec

(PodSpec). - Prefer Job-level `activeDeadlineSeconds`: - Setting

spec.activeDeadlineSeconds

on the Job is the most reliable cap for the whole Job. You can keep Pod-level ADS if you want, but it’s not necessary with

backoffLimit: 0

. - Validate exit codes in `podFailurePolicy`: - Negative codes (e.g., -1, -121) aren’t valid container exit codes in Kubernetes. Keep to 0–255 (e.g., 137 for SIGKILL, 1 for generic failure). - Double-check Slack link formatting didn’t leak into the manifest: - Replace keys that look like links with plain strings: -

<http://nvidia.com/gpu|nvidia.com/gpu>

(not

<http://nvidia.com/gpu|nvidia.com/gpu>

) -

<http://topology.kubernetes.io/zone|topology.kubernetes.io/zone>

(not

<http://topology.kubernetes.io/zone|...>

) - Annotations keys should also be plain strings (e.g., `cluster-autoscaler.kubernetes.io/safe-to-evict`). - Confirm command vs args alignment with your image: - You’re templating

args: "{{ command }}"

. That’s fine if your image ENTRYPOINT knows how to handle those args. If not, either: - Set

command: "{{ command }}"

and leave

args

out, or - Set

command: ["/bin/bash","-lc"]

and

args: ["{{ command | join(' ') }}"]

to run as a shell. - Keep retries off everywhere you want a hard stop: -

backoffLimit: 0

is good. Also ensure the Prefect flow run has no retries, and no Automations are configured to re-run on Crashed/Failed (to avoid “ghost” requeues). A cleaned-up version you can use as a reference ``` "job_configuration": { "env": "{{ env }}", "name": "{{ name }}", "labels": { "intent": "karpenter-nodepool-gpu-2x" }, "command": "{{ command }}", "namespace": "{{ namespace }}", "job_manifest": { "apiVersion": "batch/v1", "kind": "Job", "metadata": { "labels": "{{ labels }}", "namespace": "{{ namespace }}", "generateName": "{{ name }}-" }, "spec": { "backoffLimit": 0, "ttlSecondsAfterFinished": "{{ finished_job_ttl }}", "activeDeadlineSeconds": 3600, // Job-level ADS // Optional: keep completions/parallelism at Job level (usually not needed for Prefect) // "completions": 1, // "parallelism": 1, "podFailurePolicy": { "rules": [ { "action": "FailJob", "onExitCodes": { "values": [1, 9, 137], // remove negative codes "operator": "In" } }, { "action": "FailJob", "onPodConditions": [ { "type": "PodScheduled", "reason": "Unschedulable", "status": "False" } ] } ] }, "template": { "metadata": { "labels": { "app": "prefect3-DNR" }, "annotations": { "karpenter.sh/do-not-disrupt": "true", "cluster-autoscaler.kubernetes.io/safe-to-evict": "false" } }, "spec": { "restartPolicy": "Never", "terminationGracePeriodSeconds": 30, "volumes": [ { "name": "dshm", "emptyDir": { "medium": "Memory" } } ], "containers": [ { "name": "prefect-job", "image": "{{ image }}", "imagePullPolicy": "{{ image_pull_policy }}", "env": "{{ env }}", "args": "{{ command }}", // or set as

command

depending on your image

"resources": { "limits": { "cpu": "{{ limit_cpu }}m", "memory": "{{ limit_memory }}Gi", "nvidia.com/gpu": 1 }, "requests": { "cpu": "{{ requests_cpu }}m", "memory": "{{ requests_memory }}Gi", "nvidia.com/gpu": 1 } }, "volumeMounts": [ { "name": "dshm", "mountPath": "/dev/shm" } ] } ], "tolerations": "{{ tolerations }}", "nodeSelector": "{{ node_selector }}", "topologySpreadConstraints": [ { "maxSkew": 1, "topologyKey": "topology.kubernetes.io/zone", "labelSelector": { "matchLabels": { "app": "prefect3-DNR" } }, "whenUnsatisfiable": "ScheduleAnyway" } ], "serviceAccountName": "{{ service_account_name }}" } } } }, "stream_output": "{{ stream_output }}", "cluster_config": "{{ cluster_config }}", "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}", "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}" } ``` Why these changes matter for your symptoms - Job-level ADS + backoffLimit:0 ensures Kubernetes hard-stops the run and does not retry it itself. - Correct placement of

podFailurePolicy

ensures k8s “FailJob” behavior works instead of being ignored by the API. - Removing negative exit codes prevents the policy from being rejected or silently not matching. - Avoiding Automations/retries prevents Prefect from resubmitting on Crashed. - Using the right command/args pairing prevents “stuck” containers that never actually execute the Prefect command. If you still see the observer “crash and ghost requeue” after these changes, please share: - Worker and observer logs around the time the Job is killed (look for messages about marking runs as crashed or rescheduling) - The flow run’s state history from the UI (to see what’s re-queueing it) - Kubernetes Event messages for the Job/Pod (to confirm ADS vs OOM vs eviction) Helpful references - Prefect Kubernetes collection: prefect-kubernetes docs - Kubernetes Job podFailurePolicy: Pod failure policy - Topology spread constraints: Topology spread constraints Happy to iterate on this with your exact image/ENTRYPOINT and the Prefect flow retry/automation settings if you can share those.

Great questions — here’s a straight answer to each, plus how to stop the unexpected respawns you’re seeing. - Yes: it’s better to set activeDeadlineSeconds at the Job level than the Pod level - Job-level ADS caps the entire Job lifetime (all pods for that Job). It’s the clearest way to enforce a hard wall-clock limit and guarantees the Job is marked Failed after the deadline. - Pod-level ADS only caps each individual Pod. With backoffLimit > 0 it can lead to multiple short-lived Pods; with backoffLimit: 0 it usually behaves similarly, but Job-level ADS is still the more predictable choice for “one-and-done” runs. - Why you’re seeing “another pod” even with backoffLimit: 0 and no flow retries - If your Job truly has backoffLimit: 0 and there are no flow retries, Kubernetes itself should not create a new Pod after a failure. When a second Pod appears, it’s almost always Prefect resubmitting the same flow run. - The two most common Prefect-side causes: 1) SIGTERM reschedule behavior is active. Prefect 3’s flow engine can reschedule the run on SIGTERM (e.g., eviction, certain deadline stops) when the env var

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR

is set to

reschedule

. The Kubernetes worker often enables this when

backoffLimit: 0

so evictions don’t mark runs as crashed. 2) An Automation is firing (e.g., “On Crashed → Run flow”) or a manual retry was triggered from the UI/API. - Less common but worth checking: - The original Job spec mis-placed fields (e.g.,

podFailurePolicy

under pod spec instead of job spec) so Kubernetes ignored them, and the controller retried by default (backoffLimit default is 6). You’ve got

backoffLimit: 0

in the right place, but make sure other Job fields (e.g.,

completions

,

parallelism

,

podFailurePolicy

) aren’t accidentally placed under

template.spec

where they’ll be ignored. - Why Prefect is spinning up the Job again (and how to stop it) - Prefect (server/worker) will create a new Job for the same flow run when: - The flow run transitions to a retry-like state (e.g., rescheduled on SIGTERM), OR - An Automation or manual action re-queues it. - To prevent Prefect from resubmitting on infra termination: 1) Explicitly disable reschedule-on-SIGTERM inside the container:

env:
       - name: PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR
         value: "crash"

2) Keep

backoffLimit: 0

on the Job and

restartPolicy: Never

in the pod template. 3) Ensure the flow has no retries and no Automations are configured to re-run on Crashed/Failed. 4) Prefer Job-level

activeDeadlineSeconds

(and keep

terminationGracePeriodSeconds

short, e.g., 30) so kubelet escalates to SIGKILL and the Job is marked Failed decisively. Quick verification checklist - Confirm on a live run: -

kubectl get job <name> -o yaml

shows

spec.backoffLimit: 0

and

spec.activeDeadlineSeconds

present at the Job level. - The Pod’s env contains (or not)

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR

. If it’s

reschedule

, set it to

crash

. - In the Prefect UI, check the flow run’s state timeline after the kill: - If you see “AwaitingRetry” or “Scheduled” after termination, reschedule-on-SIGTERM or an Automation is at play. - If a second Pod appears, run: -

kubectl describe job <name>

to see controller events (was a new Pod created by K8s?) - Worker/observer logs around the failure (look for messages about reschedule, crash marking, or submitting infrastructure again) If you want, share: - One full Job YAML from

kubectl get job -o yaml

for a failing run - The flow run state timeline (UI -> States) - The Pod env (or your templated env block) … and I can confirm exactly which mechanism is re-queuing. References - Prefect Kubernetes collection: prefect-kubernetes docs

- Kubernetes Job concepts: Kubernetes Job docs (termination, deadlines, backoff) - Topology spread constraints: Topology spread constraints

Short answers first: - Set it where the flow actually runs: inside the job pod’s environment. Setting it on the Prefect server does nothing; setting it on the worker Deployment only helps if your worker template propagates it into the job pod’s

env

. - Fixing the misplaced fields is still important, but it will not, by itself, stop Prefect from re-running a flow if SIGTERM-reschedule or an Automation is the cause. Details and what to do Where to set PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR - Set it in the job pod’s environment (the container that runs your flow): - In your work pool’s job template, under the container’s `env`:

"env": [
      { "name": "PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR", "value": "crash" }
      // ...your other env vars
    ]

- Or embed it in the image itself (ENV line), but the job template is usually easiest. - Verifying: for a running pod, check

kubectl describe pod <pod>

or

kubectl exec <pod> -- printenv | grep PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR

. - Important: putting it on the Prefect server won’t propagate to flow pods. Putting it on the worker Deployment also won’t propagate unless your worker’s job template copies that env into the flow pod (your template already has an

env

block you can use). Will fixing podFailurePolicy placement stop the re-run? - Not by itself. The “re-run” you’re seeing is not caused by Kubernetes if `backoffLimit: 0`; it’s almost always: - SIGTERM reschedule logic in the flow process, or - A Prefect Automation (e.g., “On Crashed → Run flow”), or - A manual retry. - You should still fix the spec placement so Kubernetes actually honors your intent: - Move

podFailurePolicy

to

job_manifest.spec.podFailurePolicy

(JobSpec), not

template.spec

. - Clean up `onExitCodes`: - Remove negative codes (invalid for containers). - Use 137 (SIGKILL/OOM-kill) and 143 (SIGTERM) if you want to explicitly “FailJob” on those. - Example:

"podFailurePolicy": {
        "rules": [
          { "action": "FailJob", "onExitCodes": { "operator": "In", "values": [1, 137, 143] } },
          { "action": "FailJob", "onPodConditions": [ { "type": "PodScheduled", "reason": "Unschedulable", "status": "False" } ] }
        ]
      }

- Keep

backoffLimit: 0

at

job_manifest.spec

and

restartPolicy: Never

in

template.spec

. To definitively stop unexpected respawns - In the job pod env, set:

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash

- Ensure the flow has zero retries and there are no Automations that re-run on Crashed/Failed. - Prefer Job-level ADS and short grace: -

job_manifest.spec.activeDeadlineSeconds: 3600

-

template.spec.terminationGracePeriodSeconds: 30

- Verify behavior on a test run: - After the pod is killed by ADS, the flow run state should go to Crashed/Failed and stay there. - You should not see another Job/Pod unless a manual retry or Automation fires. If you still see a new Job/Pod after these changes: - Check the flow run state timeline in the UI (does it briefly go to AwaitingRetry/Scheduled?). - Check the pod env actually contains

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash

. - Check worker/observer logs around the failure for any “reschedule” or “submitting infrastructure” messages. K8s references - Pod failure policy: Kubernetes: Pod failure policy - Jobs and deadlines: Kubernetes: Jobs

Yes — you can set this as a default in your work pool template so every job/pod launched by the pool gets it. Where to add the env default in your work pool template - In your variables schema, add a default for the

env

object that includes the key you want. Since your job manifest already uses

"env": "{{ env }}"

at both the job and container level, this default will be injected into the pod’s container

env

automatically. Example edit to your variables block:

"variables": {
  "type": "object",
  "properties": {
    "env": {
      "type": "object",
      "title": "Environment Variables",
      "description": "Environment variables to set when starting a flow run.",
      "additionalProperties": { "type": "string" },
      "default": {
        "PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR": "crash"
      }
    },
    ...

- You can still add more env vars per-deployment/run; they’ll merge with this default. If a deployment explicitly sets

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR

, that value will override the default. Important: setting this on the Prefect server or just on the worker Deployment will not propagate into the flow pod unless it’s included in the job template

env

. The above change ensures it’s always present. About PodFailurePolicy - It does work — but only if: - It’s placed under

job_manifest.spec.podFailurePolicy

(JobSpec), not under

template.spec

(PodSpec). - You’re on a Kubernetes version that supports it (generally 1.25+). - The rules actually match the pod/container termination conditions (e.g., exit codes 1/137/143; “Unschedulable” pod condition). - What it does: it tells the Job controller how to treat certain failures (e.g., “FailJob” immediately on specific exit codes or unschedulable pods). With

backoffLimit: 0

, the practical effect is “mark the Job Failed right away,” which is what you want. - What it does not do: it does not stop Prefect from resubmitting a flow run if the flow process inside the pod handled SIGTERM in “reschedule” mode or if you have an Automation retrying on Crashed. PodFailurePolicy influences Kubernetes’ Job status; Prefect will see the Job as Failed and mark the flow run Crashed unless the flow run already put itself into an AwaitingRetry/Scheduled state (e.g., from

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=reschedule

). So, to stop the unexpected respawns: - Add the default env above to force “crash” behavior: -

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash

- Keep

backoffLimit: 0

,

restartPolicy: Never

, and prefer

spec.activeDeadlineSeconds

at the Job level. - Fix the placement of

podFailurePolicy

to

job_manifest.spec.podFailurePolicy

and use valid exit codes (remove negatives). - Ensure there are no Prefect Automations that re-run on Crashed/Failed for these deployments. If you want, paste your current

job_manifest

section again and I’ll mark up exactly where to move

podFailurePolicy

and confirm ADS/backoff placement.

Thanks for the detailed timeline — that makes the behavior clear. What you’re seeing matches Prefect 3’s current k8s observer/worker design under API outage: - The Kubernetes observer is event-driven and disables periodic scanning. If the Prefect API is unreachable at the moment a Job/Pod failure is handled, the observer’s state proposal fails and there is no built-in retry/backfill later. The run can remain stuck in Submitting/Pending even though the k8s Job is definitively Failed. - There’s a short WAIT-loop retry only when the API tells us to “WAIT” (or for a 5s InfraPending proposal), but there’s no retry/backoff when the API call itself 500s or times out. Those exceptions are logged at debug and the handler exits. Given your specific sequence (API overloaded → pod’s first API call 500s → pod exits 1 → Job backoffLimitExceeded → observer also can’t reach API), here’s how to recover and prevent repeats. Immediate recovery options - Manually set stuck flow runs to Crashed/Failed: - From the UI: set state → Crashed with a message like “Manual recovery after API outage” - Or via a quick script:

from prefect import get_client
    from prefect.states import Crashed

    async def fix(flow_run_id):
        async with get_client() as client:
            await client.set_flow_run_state(
                flow_run_id=flow_run_id,
                state=Crashed(message="Manual recovery after API outage"),
                force=True,
            )

- Restart the k8s observer (forces a fresh watch and re-diagnosis of currently present objects; it won’t backfill deleted jobs/pods): -

kubectl rollout restart deployment/prefect-kubernetes-observer -n <namespace>

- If a failed pod is still around, deleting it won’t change this specific run’s state (since the Job already failed), but it can help the observer re-process active objects on restart. Hardening to avoid future stuck runs during API blips - Ensure the flow pod does not reschedule on SIGTERM unintentionally - Set in your work pool template default env:

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash

- Keep Kubernetes in charge of hard stops and keep retries off -

spec.backoffLimit: 0

(Job) -

spec.activeDeadlineSeconds

(Job) for a hard wall clock limit -

template.spec.restartPolicy: Never

- No flow-level retries; disable any Automations that re-run on Crashed for these flows - Fix the job manifest field placement so k8s policies actually apply -

podFailurePolicy

must be under

job_manifest.spec.podFailurePolicy

- Use valid exit codes only (e.g., 1, 137, 143). Negative codes are not valid container exit codes - Reduce observer/API pressure during spikes - If you have many namespaces or a large volume of events, tune the observer concurrency: - Increase startup event concurrency to speed up initial scans: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=20

(or similar) - Consider scoping observed namespaces to only those in use - Ensure Prefect API has sufficient resources and connection pooling to handle surges (especially around batch failures) What to expect and what not to expect - The observer will not retroactively reconcile and correct states for failures that happened while the API was down; it does not periodically rescan historical k8s events or retry failed state proposals. - After the API recovers, the observer will process new watch events going forward and will rescan currently present Jobs/Pods on startup — but it will not backfill runs that already missed proposals while the API was unavailable. If you share a representative flow_run_id that’s stuck plus the observer logs around that timestamp, I can confirm whether it hit the “API unavailable during propose_state” path and suggest any additional tuning specific to your environment.

I hear you — at scale, losing the observer’s state proposals during an API spike will leave pods Failed in k8s but “Running” or “Submitting” in the UI. There are two things to do in parallel: make the observer resilient so it restarts automatically and reduces API load, and add a lightweight reconciler so missed updates are corrected even if the observer can’t reach the API in the moment. 1) Make the observer resilient (so you don’t need to manually rollout restart) - Add liveness/readiness probes to the observer Deployment so kubelet restarts it automatically if it crashes or hangs:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: prefect-kubernetes-observer
spec:
  template:
    spec:
      containers:
      - name: observer
        image: your-registry/prefect-observer:tag
        # resources sized for your scale
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2"
            memory: "2Gi"
        env:
        - name: PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED
          value: "true"
        - name: PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES
          value: "research-prefect3"   # comma-separated to shard load
        - name: PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY
          value: "5"                   # lower to reduce API spikes; raise if startup is slow
        - name: PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS
          value: "false"               # disables per-pod event replication; big API saver
        # Optionally: filter only jobs created by your work pool
        # - name: PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ADDITIONAL_LABEL_FILTERS
        #   value: "<http://prefect.io/work-pool-id=<id>|prefect.io/work-pool-id=<id>>"
        livenessProbe:
          exec:
            command: ["sh","-c","pgrep -f prefect_kubernetes.observer >/dev/null"]
          initialDelaySeconds: 20
          periodSeconds: 10
          failureThreshold: 3
        readinessProbe:
          exec:
            command: ["sh","-c","pgrep -f prefect_kubernetes.observer >/dev/null"]
          initialDelaySeconds: 10
          periodSeconds: 10
          failureThreshold: 3

- Scope the observer by namespace(s) and/or labels; this has the biggest impact on API load. - Consider running multiple observer Deployments, each scoped to a distinct namespace set, to shard the work. - On the server side, increase event batching to reduce DB pressure (if you self-host Prefect Server): - PREFECT_SERVER_SERVICES_EVENT_PERSISTER_BATCH_SIZE=20 (or higher) - PREFECT_SERVER_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=5 - Ensure the Prefect API deployment is horizontally scalable and the database pool/limits are sized for your peak concurrency. 2) Add a lightweight reconciler (safety net for missed proposals) Because the observer does not reconcile missed events after API outages, a small reconciler that runs every N minutes can clean up “Running”/“Submitting” runs whose k8s Jobs are already Failed. You can run this as a Kubernetes CronJob. Example skeleton (Python): ``` import asyncio import os import json from kubernetes import client, config from prefect import get_client from prefect.states import Crashed NAMESPACE = os.getenv("NAMESPACE", "research-prefect3") LABEL_SELECTOR = os.getenv("LABEL_SELECTOR", "app=prefect3-DNR") # match your labels async def main(): # K8s client if os.getenv("KUBERNETES_SERVICE_HOST"): config.load_incluster_config() else: config.load_kube_config() batch = client.BatchV1Api() core = client.CoreV1Api() # Prefect client async with get_client() as pc: jobs = batch.list_namespaced_job(namespace=NAMESPACE, label_selector=LABEL_SELECTOR).items for job in jobs: # Expect flow run id label on job/pod from your template fr_id = job.metadata.labels.get("prefect.io/flow-run-id") if not fr_id: continue

# If Job failed (BackoffLimitExceeded or Failed condition), mark run Crashed if still Running/Pending failed = any( c.type == "Failed" and c.status == "True" for c in (job.status.conditions or []) ) if not failed: continue try: fr = await pc.read_flow_run(fr_id) except Exception: continue if fr.state and fr.state.type.value in {"RUNNING", "PENDING", "SCHEDULED", "SUBMITTED"}: try: await pc.set_flow_run_state( flow_run_id=fr_id, state=Crashed(message="Reconciled after k8s Job failed (observer/API outage)"), force=True, ) except Exception as e: # Log and continue; retry next run print(f"Failed to set state for {fr_id}: {e}") if name == "__main__": asyncio.run(main()) ``` - Run this as a CronJob every few minutes to guarantee eventual correctness if the observer/API hiccups at failure time. - Tune the label selector and the “terminal” conditions to your environment (e.g., also look at Pod container termination reasons like OOMKilled). 3) Double down on preventing unnecessary re-runs and reducing event volume - Keep k8s in charge of hard stops: Job.spec.activeDeadlineSeconds with backoffLimit: 0; Pod.spec.restartPolicy: Never. - Force “crash” on SIGTERM to avoid reschedules: - Set in your work pool template env defaults: PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash - Fix and validate JobSpec fields (so k8s behaviors actually apply): - Move podFailurePolicy to job_manifest.spec.podFailurePolicy; remove negative exit codes; include 1, 137, 143 as needed. - Consider disabling pod event replication by the observer (PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false). You’ll still get Job-level detection for success/failure/crash with far fewer API calls. 4) If the Prefect API is your bottleneck - Scale API replicas and CPU/memory; ensure DB connection pool and max connections are sized for load. - Ensure workers/observers use connection pooling (they do by default) and avoid thundering herds by: - Lowering PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY if API is fragile at startup - Sharding observers by namespace/labels to limit events per observer To your direct question: yes, you can avoid manual kubectl rollout restart by giving the observer proper liveness/readiness probes and resource headroom. K8s will restart the container automatically when it crashes or becomes unhealthy. If you can share: - How many concurrent k8s Jobs/pods you run at peak - The namespaces/labels the observer is watching - Prefect API replica/DB sizing …I can suggest concrete numbers for observer concurrency, server batch sizes, and whether it’s better to run multiple observers sharded by namespace. If you’d like, I can also package the reconciler above as a Helm template/CronJob for your cluster.

I’m really sorry this has been so painful. Running ~1,000 concurrent GPU jobs is serious scale, and if the observer can’t keep up or the API has short outages, you’ll see exactly what you’re seeing: pods die while runs stay “Running/Submitting” because the state proposal was missed. Two direct answers to your questions: - Prefect does not hard-code the observer; it’s fully configurable via environment variables. If you don’t see it exposed in your chart, you can still set the env vars on the worker/observer deployment. The observer reads Prefect settings at process startup. - You do not have to live with manual kubectl restarts; you can harden the observer (probes/resources) and aggressively reduce its event volume. You can also disable high-churn features (pod-event replication) or even disable the observer entirely and run a tiny reconciler CronJob for eventual correctness. Concrete steps I recommend for your scale A) Reduce observer load and scope it tightly - Set these on the worker/observer Deployment (env for the container that runs the Prefect worker/observer):

# Enable the observer but scope it tightly
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true
# Only watch the namespaces you actually use (shard if needed)
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=research-prefect3
# Big win: stop replicating every Pod event; stick to Job-level outcomes
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false
# Be gentle during startup; avoid thundering herds against the API
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3
# Optional: only watch pods/jobs from this pool/team
# (<http://prefect.io/work-pool-id|prefect.io/work-pool-id> is automatically labeled by workers; use it if available)
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ADDITIONAL_LABEL_FILTERS=<http://prefect.io/work-pool-id=<your-pool-id|prefect.io/work-pool-id=<your-pool-id>>
# Optional: if log forwarding on crash is expensive for your cluster
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_FORWARD_CRASHED_RUN_LOGS=false

- Shard observers: run multiple worker+observer deployments, each filtered to a distinct namespace or label (e.g., 4 observers each watching ~250 jobs/nodes). This flattens spikes and isolates failures. - Add liveness/readiness probes and give the observer more CPU/memory so kubelet auto-restarts it if it stalls:

livenessProbe:
  exec: { command: ["sh","-c","pgrep -f prefect_kubernetes.observer >/dev/null"] }
  initialDelaySeconds: 20
  periodSeconds: 10
readinessProbe:
  exec: { command: ["sh","-c","pgrep -f prefect_kubernetes.observer >/dev/null"] }
  initialDelaySeconds: 10
  periodSeconds: 10
resources:
  requests: { cpu: "500m", memory: "512Mi" }
  limits:   { cpu: "2",    memory: "2Gi"   }

B) Prevent unintended re-runs and keep k8s in charge of hard stops - Force “crash” on SIGTERM so infra terminations don’t reschedule the same run:

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash

- In your JobSpec: -

spec.backoffLimit: 0

-

spec.activeDeadlineSeconds: <your wall clock limit>

(Job-level) -

template.spec.restartPolicy: Never

- Fix placement of

podFailurePolicy

under

job_manifest.spec.podFailurePolicy

and only use valid codes (e.g., 1, 137, 143). C) Add an eventual-consistency safety net (CronJob reconciler) Because the observer won’t backfill state proposals lost during API blips, run a tiny reconciler every few minutes that: - Looks for k8s Jobs in Failed/Succeeded - Reads the corresponding flow run - Sets the final Prefect state if the run is still “Running/Submitting” Skeleton you can adapt (earlier in the thread I posted a full example): - If Job Failed → set flow run to Crashed (force) - If Job Succeeded → set flow run to Completed (force) This guarantees UI correctness even if the observer or API hiccuped at failure time. D) Server-side smoothing (if you self-host Server) - Batch events more aggressively to reduce DB churn: -

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_BATCH_SIZE=20

-

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=5

- Keep API replicas, gunicorn/uvicorn workers, and DB pool sized for your concurrency pattern. 50% CPU isn’t the whole story: bursts, connection exhaustion, or event write amplification will still cause 5xx without maxing CPU. Where to put these env vars - On the Deployment that runs your Prefect Kubernetes worker (which also runs the observer when enabled). If you’ve split an “observer-only” Deployment, put the same env vars there. - If you use a Helm chart that doesn’t expose these explicitly, add them under the container’s

env:

list in your values.yaml (or via

extraEnv:

if your chart supports it). They’re standard Prefect settings and are picked up on process start. Reality check at your scale (1001 concurrent jobs) - You will almost certainly need: - Multiple worker+observer pods, sharded by namespace/label - Pod event replication disabled - Tight namespace/label filters - A reconciler to guarantee final-state correctness when the observer/API misses windows - With that in place, “pods crash but run still shows Running” should disappear even if there are transient API errors. I know you’re frustrated — the above isn’t hand-waving; it’s the set of levers Prefect 3 exposes for high-volume k8s environments. If you can share: - How many namespaces you’re watching today - Whether you can shard by work pool/label - A short window of observer and API logs during a surge …I’ll map out an exact sharding plan (e.g., 4 observers × 250 jobs, concrete env values) and give you a ready-to-apply values.yaml snippet. Docs for reference: - Kubernetes observer/config settings: prefect-kubernetes docs - Kubernetes Jobs/Backoff/ADS: Kubernetes Job docs

First, where the observer “lives” - It runs inside the Kubernetes worker process (same pod). It is NOT scheduled on the same node as your jobs and it is NOT one-per-job. It watches your cluster via the Kubernetes API from the worker pod(s). - With a single namespace, you should scope the observer to that namespace so it does not watch cluster-wide. What to change — short, prioritized, and exact Apply in this order so you can test incrementally. 1) Stop unintended resubmissions from flow pods (highest impact) - Add this to your work pool template’s env defaults so every job pod gets it:

"variables": {
  "properties": {
    "env": {
      "type": "object",
      "default": {
        "PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR": "crash"
      }
    }
  }
}

- Effect: When Kubernetes sends SIGTERM (eviction, deadline, etc.), the flow will not auto-reschedule; it will crash/terminate cleanly. 2) Reduce observer load and scope it to your single namespace (very high impact) Set these env vars on the Deployment that runs your Prefect Kubernetes worker (the one that hosts the observer):

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-single-namespace>
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

- Effect: - Only watches your one namespace (not cluster-wide). - Disables per-pod event replication (big API saver) while still detecting Job success/failure/crash. - Avoids API thundering herd at observer startup. 3) Ensure the Job spec is authoritative (consistent hard-stop behavior) - Put these in

job_manifest.spec

(JobSpec), not under

template.spec

(PodSpec): -

backoffLimit: 0

-

activeDeadlineSeconds: <your wall clock limit>

-

podFailurePolicy: { ... valid exit codes only (e.g., 1, 137, 143) ... }

- Keep these in `job_manifest.spec.template.spec`: -

restartPolicy: "Never"

-

terminationGracePeriodSeconds: 30

- Effect: Kubernetes enforces hard time limits and “one-and-done” failures; Prefect won’t see k8s-driven retries. 4) Run exactly one observer-enabled worker; disable on any others (prevents duplication) - If you run multiple worker replicas, leave one with: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true

- Set the rest to: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

- Effect: Eliminates duplicate watchers and redundant API calls while still allowing many workers to submit jobs. 5) Optional but recommended for eventual correctness - Add a small CronJob “reconciler” to mark runs Crashed/Completed if the observer/API missed the window during a blip. This guarantees UI correctness at scale even with transient failures. Notes - Observer config is not “hidden” — it’s standard Prefect settings read at process start. Add those env vars to your worker Deployment/container spec. - With 1000+ concurrent jobs, #2 and #4 are crucial. Most clusters stabilize dramatically after disabling pod-event replication and scoping to the single namespace. If you’d like, paste the worker Deployment env section and I’ll mark up exactly where to add these, and I can double-check your job_manifest for spec placement. Reference docs: prefect-kubernetes docs

Totally understand. That stack trace means the API tried to write to a WebSocket after it was already closed. At your scale, this is usually caused by a client-side or load-balancer/ingress idle timeout closing the socket under load; the server then races and attempts a send. The 85% memory note is a red herring here — it can worsen GC pauses and latency, but the key symptom is a closed WebSocket. Where the observer runs - It runs inside the Prefect Kubernetes worker pod (same process). It is not co-scheduled with your jobs. It watches via the Kubernetes API from wherever the worker pod is scheduled. Highest-priority, short list with exact changes Apply in order and test between steps. 1) Slash event volume from Kubernetes (biggest win) - On your worker Deployment (the one that runs the observer), set:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-single-namespace>
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

- Why: Scopes to your single namespace, stops per-pod event replication (huge event reduction), avoids API thundering herd at startup. This directly reduces pressure on your API and UI WebSockets. 2) Prevent unintended resubmissions from pods - In your work pool template defaults, add this to the env object so every job pod gets it:

"PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR": "crash"

- Why: Ensures infra terminations (ADS, evictions) don’t trigger a reschedule loop that spins up more jobs and more events. 3) Fix Job spec so Kubernetes is authoritative on hard stops - In JobSpec (job_manifest.spec), not in template.spec: - backoffLimit: 0 - activeDeadlineSeconds: <your hard wall time> - podFailurePolicy with only valid codes (e.g., 1, 137, 143) - In PodSpec (job_manifest.spec.template.spec): - restartPolicy: Never - terminationGracePeriodSeconds: 30 - Why: Ensures “one-and-done” behavior, no k8s retries, deterministic termination. 4) Stabilize WebSockets at the ingress/load balancer (often the real cause of that error) - If using NGINX Ingress, add annotations on your API ingress: - nginx.ingress.kubernetes.io/proxy-read-timeout: "600" - nginx.ingress.kubernetes.io/proxy-send-timeout: "600" - nginx.ingress.kubernetes.io/enable-websocket: "true" - If using AWS ALB/NLB or another LB, increase idle timeout to 300–600s and ensure WebSocket/HTTP2 is enabled. - Why: Prevents idle closures under load; stops the “send after close” error at the source. 5) Run exactly one observer-enabled instance; disable observers on any extra workers - Keep one worker/observer with: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true - Set all other worker replicas to: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false - Why: Eliminates duplicate watchers and cuts event fan-out without reducing submission capacity. 6) Optional but highly recommended: eventual-consistency reconciler - Run a small CronJob every few minutes: - If a k8s Job is Failed/Succeeded and the Prefect run is still Running/Submitting, set it to Crashed/Completed (force). - Why: Guarantees UI correctness even if an outage hits right when the observer should have proposed final state. If your API is behind a reverse proxy, step 4 is critical — that specific Starlette RuntimeError is what you see when a WebSocket is closed by the client/LB and the server thread tries to send anyway. Want me to turn these into a ready-to-paste values.yaml patch? If you share: - Your worker Deployment env section - Your API ingress annotations (or LB type)

I’ll annotate exactly where to add each setting. References: prefect-kubernetes docs, Kubernetes Job docs

Got it — thanks for clarifying your network path. What’s going on - The stack trace you shared is from the API trying to send on a WebSocket that the client has already closed. At your scale, the most common cause is an idle timeout somewhere on the external path used by browsers (UI) — typically the ALB or a reverse proxy in front of the API. - Your workers/observers inside the cluster talk to the API via ClusterIP, which generally won’t hit the ALB/NLB. So: - External path (UI → ALB → API): most likely to cause these “send after close” errors. - Internal path (pods → ClusterIP → API): generally not affected by ALB/NLB timeouts. What to change for AWS ALB/NLB and reverse proxy Prioritized, with exact knobs. Apply to the external/UI path first; this is where your error almost certainly originates. 1) ALB (external path for the UI) - Increase the ALB idle timeout to 600 seconds (or higher, up to 4000s). - If you use the AWS Load Balancer Controller with an Ingress, add:

<http://alb.ingress.kubernetes.io/idle-timeout|alb.ingress.kubernetes.io/idle-timeout>: "600"
<http://alb.ingress.kubernetes.io/backend-protocol-version|alb.ingress.kubernetes.io/backend-protocol-version>: HTTP1

- If you manage the ALB directly, set the Target Group attribute “Connection idle timeout” to 600s in the console or via Terraform/CloudFormation. - Why: Prevents the ALB from closing idle WebSockets that the UI keeps open to receive live event updates. Docs: - AWS ALB idle timeout: https://docs.aws.amazon.com/elasticloadbalancing/latest/application/application-load-balancers.html#connection-idle-timeout 2) Reverse proxy between ALB and API (if present, e.g., NGINX Ingress in front of the API) - Add these annotations to the API’s Ingress to keep WebSockets alive and prevent upstream timeouts:

<http://nginx.ingress.kubernetes.io/enable-websocket|nginx.ingress.kubernetes.io/enable-websocket>: "true"
<http://nginx.ingress.kubernetes.io/proxy-read-timeout|nginx.ingress.kubernetes.io/proxy-read-timeout>: "600"
<http://nginx.ingress.kubernetes.io/proxy-send-timeout|nginx.ingress.kubernetes.io/proxy-send-timeout>: "600"

- If you’re not using NGINX Ingress, set equivalent keepalive/idle timeouts in your proxy (Envoy/Istio, HAProxy, Traefik). - Why: Even if the ALB is tuned, an aggressive proxy timeout in front of the API will still close the socket early. Docs: - NGINX Ingress annotations: https://kubernetes.github.io/ingress-nginx/user-guide/nginx-configuration/annotations/ 3) NLB specifics (if you actually route WebSockets through an NLB) - NLB is TCP pass-through and generally fine for WebSockets, but idle timeouts can still occur (defaults can be shorter than you expect depending on config). If you’re using NLB → NGINX Ingress, the NLB won’t fix idle closures at the NGINX layer — you still need the NGINX timeouts above. - If you can, keep WebSockets on ALB HTTP/1.1 instead of NLB unless you have a specific need for TCP pass-through. 4) Confirm external vs internal paths - Internal (workers/observers → API): If these calls go Service ClusterIP → kube-proxy → API Pods, ALB/NLB settings won’t apply and shouldn’t cause this error. - External (UI in browser → ALB → API): This is where the WebSocket closure almost certainly occurs. - You can verify by: - Checking ALB access logs around the timestamps of the error for WebSocket connections closing near the idle timeout. - Watching Ingress logs for upstream timeouts (e.g., 504s/499s). 5) Optional server-side smoothing - If you see frequent reconnects under load even after the above: - Make sure your API pods have enough CPU/memory headroom to avoid long GC or event serialization pauses. - Scale API replicas to spread bursty WebSocket traffic (you have 8; make sure session routing or sticky behavior isn’t interfering — WebSockets stick to one backend by design). - If you run a proxy next to the API (sidecar/ingress), enable HTTP/1.1 upgrade and keepalive there too. Important note

- These WebSocket errors do not directly cause “pods show as running in the UI” — that symptom is from missed state proposals (observer/API outage timing). Stabilizing WebSockets will quiet errors and help UI live updates, but you should still apply the observer-scope and reconciler steps we discussed to guarantee final state correctness at your scale. If you can share: - Whether the API is fronted by ALB → NGINX Ingress → Service, or ALB directly to Service - Any current ALB Ingress annotations and NGINX annotations …I’ll give you a ready-to-paste patch with the exact annotations for your setup.

Short answer: yes — tuning the ALB/ingress WebSocket timeouts primarily fixes UI/live-update symptoms. It won’t fix missed state proposals or pods showing “Running” in the UI when they’ve already died. What ALB/ingress WebSocket tuning actually affects - Fixes: - UI live updates (event stream) dropping and reconnecting - Starlette “send after close” errors and related log noise - Any external consumers of the events WebSocket (custom dashboards, etc.) - Does not fix: - Worker/observer → API calls (these go over ClusterIP inside the cluster) - State proposals from the observer (Crash/Completed) or flow runs stuck in Submitting/Running - Job cancellations, retries, or infra behavior When it helps beyond the UI - If any internal/external clients connect to the events WebSocket through the ALB (e.g., remote workers outside the cluster, custom services), tuning ALB idle timeouts also helps their stability. - If your workers/observers are ever routed through the ALB (not your current setup), it would also improve cancellation WebSocket stability; otherwise, no effect. What still needs to be done for correctness at scale - Scope and quiet the observer to reduce API churn: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace> - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3 - Prevent unintended resubmissions from pods: - PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash in the job pod env - Ensure Job-level hard-stop behavior: - Job.spec.backoffLimit: 0, Job.spec.activeDeadlineSeconds, Pod.restartPolicy: Never - Run only one observer-enabled instance; disable on extra workers - Add a small reconciler CronJob to force-finalize runs when k8s Job is Failed/Succeeded but Prefect missed the state transition during an API blip So: tune ALB/ingress to stop the WebSocket errors in the UI, but rely on the observer tuning + reconciler to fix the core “pods died but UI says running” issue.

Short answer: not by themselves. What those k8s settings do - Job.spec.backoffLimit: 0 — prevents Kubernetes from retrying the Job - Job.spec.activeDeadlineSeconds — hard-kills the pod/marks Job Failed at the deadline - Pod.restartPolicy: Never — prevents pod restarts within the Job These make the Job outcome deterministic, but they do not push state to Prefect. Prefect only updates the flow run state when the Kubernetes observer (in the worker process) or a fallback process can reach the Prefect API and propose the final state. What you’ll see when the API/observer is healthy - Job Succeeded → Prefect state: Completed - Job Failed (non‑zero exit, OOMKilled, ADS, eviction) → Prefect state: Crashed - Prefect-issued cancel (dashboard/API) → worker deletes Job → Prefect state: Cancelled - Note: ADS firing is not a “cancel” and will not produce a Cancelled state; it will produce Crashed. Why runs still look “Running/Submitting” - If the observer can’t reach the Prefect API when the Job fails, the state proposal is missed. Your k8s settings don’t change that behavior. What to add so states are correct, even during brief API blips - Keep the k8s settings you listed (they’re necessary). - Ensure one healthy, scoped observer: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace> - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3 - Prevent infra-triggered resubmissions: - In the job pod env: PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash - Add a tiny reconciler CronJob that: - Scans for k8s Jobs in Failed/Succeeded - Forces Prefect states to Crashed/Completed if the run is still Running/Submitting - This guarantees eventual correctness regardless of transient API/observer outages Reference: prefect-kubernetes docs

Here’s exactly what each setting does, why you’d use it, and the trade-offs. - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true - What it does: Turns the Kubernetes observer on inside the worker process. The observer watches Jobs/Pods and proposes final states (e.g., Crashed/Completed) to the Prefect API. - Why use it: Without the observer, Prefect won’t automatically transition flow runs when the k8s Job finishes; you’d need a reconciler or manual updates. - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace> - What it does: Scopes the observer to a comma-separated list of namespaces instead of watching the whole cluster. - Why use it: Dramatically reduces API calls and memory/CPU usage. With a single namespace, this is a big, safe win. - Trade-off: None functionally; you just won’t see objects outside the specified namespaces. - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false - What it does: Disables per-pod event replication and pod-level handlers. The observer will still handle Job-level success/failure/crash, but it will stop emitting pod-phase events (e.g., Pending → Running) and stop proposing “InfrastructurePending” when pods are stuck pending. - Why use it: Huge reduction in event volume and API chatter at scale. You still get correct final states from Job outcomes. - Trade-offs: - You lose pod-phase events in Prefect (fewer live updates/automations driven by pod events). - You won’t see “Awaiting infrastructure” transitions for pods stuck Pending; they’ll remain Running/Pending until they succeed/fail at the Job level. - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3 - What it does: Limits the number of concurrent API calls the observer makes during startup when it deduplicates existing objects. Default is 5; setting to 3 is gentler on the API. - Why use it: Prevents thundering-herd spikes against your API/k8s when the observer starts or restarts in a large cluster. - Trade-off: Slightly slower observer startup/initial scan, but much lower risk of overload. Where to set these - Add them as env vars on the Deployment that runs your Prefect Kubernetes worker (the observer runs inside the worker pod). If you run multiple workers, enable the observer on one (or shard by namespace) and set it to false on the others to avoid duplicate watchers. What these do not do - They don’t retroactively fix runs that already missed state proposals during an API outage. - They reduce load and missed proposals going forward; pair with a small reconciler CronJob if you want guaranteed eventual correctness. Reference: prefect-kubernetes docs

Short answers up front - The observer runs inside the Kubernetes worker pod. It is not co-scheduled with your jobs and does not “follow” a job to its node. - A job pod crashing should not crash the observer. What typically happens in your failure case is: the observer tries to propose a final state right when the Prefect API is unavailable; the handler bails and the state proposal is missed. That looks like “UI says Running but no pod”. What your log snippet shows - Multiple workers raced to submit the same flow run. The “Aborted submission… PENDING to PENDING” lines are expected when you run many workers against the same pool — the server lets only one “win” the claim; the others abort. - “Creating Kubernetes job… / Flow run submitted to infrastructure” means the winner created the Job. - “Starting flow run process… Downloading flow code…” are logs emitted from inside the flow pod entrypoint; that can only appear if a Pod actually started. If you “never had an active pod” for this run, these later logs may be from a resubmission or a different run — or the original Pod started much later than you expected. How to confirm exactly what happened for that run Run these with your namespace and the run id you pasted:

kubectl get jobs -n <ns> -l <http://prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751|prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751> -o wide
kubectl get pods -n <ns> -l <http://prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751|prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751> -o wide
kubectl describe job -n <ns> -l <http://prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751|prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751>
kubectl describe pod -n <ns> -l <http://prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751|prefect.io/flow-run-id=1d2d45f2-c45a-4ec5-9dc9-87b759a5a751>

- If nothing returns, it likely finished/failed and was GC’d (TTL), or the Job was never created (API error) — check the worker logs around “Creating Kubernetes job…” for the Job name and any API errors. - If there is a Job with Failed/BackoffLimitExceeded and Prefect still shows Running, that’s the classic “observer couldn’t reach API at failure time” case. High‑priority fixes to prevent the “Running with no pod” symptom Apply in this order; each is safe to test incrementally. 1) Run exactly one observer-enabled worker (reduce races and API churn) - On one worker pod: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true - On all other worker replicas: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false - Why: You still get full submission capacity from many workers, but only one observer watches and proposes states. 2) Scope and quiet the observer - On the observer-enabled worker, set:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace>
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

- Why: Watch only your single namespace, stop per-pod event replication (biggest API saver), and avoid thundering-herd startup. 3) Prevent infra-triggered resubmits from pods - In your work pool template’s env defaults, add:

"PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR": "crash"

- Why: Ensures ADS/evictions don’t reschedule the same run, which otherwise creates more pods/events and confusion. 4) Keep Kubernetes authoritative on hard stops (you already have most of this) - In JobSpec (job_manifest.spec): backoffLimit: 0, activeDeadlineSeconds: <limit>, podFailurePolicy with valid exit codes (1, 137, 143). - In PodSpec (job_manifest.spec.template.spec): restartPolicy: Never, terminationGracePeriodSeconds: 30. 5) Add a tiny reconciler CronJob (eventual consistency) - Every few minutes, if a k8s Job is Failed/Succeeded but the Prefect run is still Running/Submitting, force the final state (Crashed/Completed). This guarantees UI correctness even if the observer/API blipped during failure.

If you want, paste your worker Deployment env section and I’ll mark exactly where to add the observer settings. Also, if you can run the kubectl label queries for that run id, we can confirm whether a pod actually started or if the state got stuck before the Job ever created. Reference: prefect-kubernetes docs

Short answer: No. A worker’s “limit 201 jobs” only throttles how many flow runs that worker will submit/run at once. The observer in that worker is a separate watcher. By default it watches all Prefect-labeled Jobs/Pods in its configured namespaces, not just “the ones it submitted.” If you run multiple workers with the observer enabled in the same namespace, each observer will watch the same set — multiplying event traffic. What to do (short, prioritized) 1) Simplest and safest: run a single observer - On ONE worker Deployment (the “observer”): - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace> - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3 - On ALL other worker Deployments: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false - Why: Prevents N× duplicate watches and slashes API load while keeping full submission capacity. 2) If you must keep multiple observers: shard them by label - Add a stable label to every Job/Pod your work pool creates (via your job template): - In job_manifest.metadata.labels and template.metadata.labels add: - "observer-scope": "group-a" (set per worker group) - Configure each worker’s observer to only watch its group: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace> - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ADDITIONAL_LABEL_FILTERS=observer-scope=group-a - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false - Note: If your jobs already carry a

<http://prefect.io/work-pool-id|prefect.io/work-pool-id>

label, you can filter on that instead and skip adding a new label: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ADDITIONAL_LABEL_FILTERS=prefect.io/work-pool-id=&lt;pool-id> 3) Keep k8s in charge of hard stops (independent of observers) - Job.spec.backoffLimit: 0 - Job.spec.activeDeadlineSeconds: <limit> - Pod.spec.restartPolicy: Never - Pod.spec.terminationGracePeriodSeconds: 30 Why this matters - There’s no built-in “observer only watches my worker’s 201 jobs” linkage. Without scoping, every enabled observer sees every Prefect Job/Pod in the namespace, so 5 workers = 5x event traffic. Server-side dedup prevents double state changes, but the API still handles all those requests — that’s where you see overload and missed proposals. If you share how many worker replicas you run today, I can suggest an exact split (e.g., 1 observer-enabled worker + N observer-disabled workers) or a label-sharding plan with concrete env and template patches.

Short answer: ALB/ingress tweaks won’t help 500s inside your flow pods. Those pods talk to the API via ClusterIP. The observer tuning we discussed is what will actually reduce API contention and, in turn, reduce 500s seen inside pods. What will help the 500s inside pods - Run one observer (or shard by label). On one worker: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true

-

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace>

-

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false

-

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

- On all other workers: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

- Optional API load reducers: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_FORWARD_CRASHED_RUN_LOGS=false

(fewer log-fetch calls) - Server side (if self-hosting): -

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_BATCH_SIZE=20

-

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=5

Why this helps - Multiple observers in the same namespace all watch the same Jobs/Pods, multiplying event traffic and state proposals. Disabling duplicates and turning off pod-event replication cuts a large portion of API traffic, which lowers the probability of 500s for flow pods calling

read_flow_run

,

set_flow_run_state

, etc. Hardening your jobs for brief API hiccups - Add a short exponential backoff around the first few API calls in your flow entrypoint to survive transient 500s:

import anyio
from prefect import get_client

async def robust_read_flow_run(flow_run_id, attempts=6, base_delay=1):
    delay = base_delay
    for i in range(attempts):
        try:
            async with get_client() as c:
                return await c.read_flow_run(flow_run_id)
        except Exception:
            if i == attempts - 1:
                raise
            await anyio.sleep(delay)
            delay *= 2

- This doesn’t solve root cause, but it prevents immediate exits when the API blips for a few seconds. What won’t help these 500s - ALB/ingress WebSocket idle timeouts. Those affect the browser/UI event stream, not in-cluster pods talking to the API via ClusterIP. Validate the impact - Before/after: track API 5xx rate, p95/p99 latency, and observer request volume. You should see fewer 500s in pod logs once duplicate observers and pod-event replication are removed. If you want, share how many worker replicas you run today. I can give you an exact plan (e.g., 1 observer-enabled worker + N observer-disabled) and a ready-to-paste env patch for your Deployment. Docs: prefect-kubernetes docs

Short answer: yes, give the observer-enabled worker more headroom — but you shouldn’t need “huge” memory if you disable pod-event replication and scope it to your single namespace. What drives resource use - Memory - Python process + k8s client objects + small in‑memory caches for last-seen events - Optional crashed-pod log forwarding buffers (can be disabled) - Spikes at startup if there are many existing Jobs/Pods to scan - CPU - JSON (de)serialization of watch events - State proposals/log forwarding to the Prefect API Recommended starting point (1 observer watching ~1,000 Jobs in a single namespace) - If you set: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false

-

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace>

-

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

- Then start the observer-enabled worker at: - requests: cpu 1, memory 1–1.5Gi - limits: cpu 2–3, memory 2–3Gi - If you must keep pod-event replication enabled, bump memory by ~1Gi and CPU by ~1 vCPU: - requests: cpu 2, memory 2–3Gi - limits: cpu 3–4, memory 3–4Gi Additional knobs that reduce memory/IO - Disable/trim crash log forwarding if not critical: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_FORWARD_CRASHED_RUN_LOGS=false

- or lower tail:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_FORWARD_CRASHED_RUN_LOGS_TAIL_LINES=200

- Run exactly one observer-enabled worker; set

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

on all other workers to avoid N× duplicate watchers. Health checks and stability - Add probes so kubelet auto-restarts if it stalls:

livenessProbe:
  exec: { command: ["sh","-c","pgrep -f prefect_kubernetes.observer >/dev/null"] }
  initialDelaySeconds: 20
  periodSeconds: 10
readinessProbe:
  exec: { command: ["sh","-c","pgrep -f prefect_kubernetes.observer >/dev/null"] }
  initialDelaySeconds: 10
  periodSeconds: 10

How to iterate safely - Implement the env settings above (namespace scope + no pod-event replication). - Set resources to requests cpu=1, mem=1.5Gi; limits cpu=3, mem=3Gi. - Observe for a busy window: - Pod RSS (resident memory), CPU saturation, restarts - Prefect API 5xx rate and p95 latency - If steady-state RSS < 1Gi and CPU < 1 vCPU, you can ratchet down. If you see OOM/restarts or high GC, bump memory by 0.5–1Gi. If even after these changes you see pressure, shard the observer - Run two observer-enabled workers, each filtered by an additional label (or work-pool label) to watch ~500 Jobs each: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ADDITIONAL_LABEL_FILTERS=observer-scope=group-a

(and label your Jobs accordingly) Docs for reference - Prefect Kubernetes settings and observer: https://prefecthq.github.io/prefect-kubernetes

Short answer: You can run a single, vertically scaled worker that also hosts the observer — it will work if you scope/quiet the observer — but I generally recommend 1 observer-enabled worker plus a few small submission-only workers. Here’s why and how. What I recommend (balanced + resilient) - 1 observer-enabled worker (vertically scaled) - Handles all Job/Pod watching and state proposals - Settings:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=true
    PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace>
    PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false
    PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

- Resources (starting point): requests cpu=1–1.5, mem=2Gi; limits cpu=3–4, mem=3–4Gi - Add probes so kubelet restarts it if it stalls - N submission-only workers (keep your 201/run caps) - Same pool, just disable the observer on them:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

- Modest resources (e.g., cpu=0.5–1, mem=512Mi–1Gi each) Why this is better than “one big worker for everything” - Avoids single point of failure: if the one big worker restarts, both submission and observation pause; you can miss state proposals during the gap. - Handles spikes better: multiple submitters can create Jobs in parallel without multiplying observer load. - Still only one observer watching the namespace, so you avoid N× duplicate watches. If you do want exactly one worker for everything - Yes, it can work if you: - Vertically scale it as above - Set the observer env vars shown - Raise that worker’s job limit high enough (e.g., 1001) - Keep k8s hard-stop settings in your JobSpec (backoffLimit: 0, activeDeadlineSeconds, restartPolicy: Never) - Add a tiny reconciler CronJob for eventual correctness during any observer/API blips - Trade-offs: - SPOF for submission and observation - Higher chance of missing state proposals during restarts - Potential burst against the k8s API if one process submits all 1000 Jobs at once Priority order to implement (incremental testing) 1) Immediately run exactly one observer-enabled instance; disable the observer on all others. 2) On that observer-enabled worker, set:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES=<your-namespace>
   PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS=false
   PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_STARTUP_EVENT_CONCURRENCY=3

3) Prevent infra resubmits from pods: - In your work pool template env defaults:

"PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR": "crash"

4) Keep JobSpec authoritative: - backoffLimit: 0, activeDeadlineSeconds, Pod restartPolicy: Never 5) Size the observer-enabled worker (requests cpu=1–1.5, mem=2Gi; limits cpu=3–4, mem=3–4Gi). Adjust based on observed RSS/CPU. 6) Optional: add a reconciler CronJob to set runs Crashed/Completed if k8s Job is terminal but the Prefect run is still Running/Submitting. If you share how many worker replicas you have now, I can give a concrete “X submitters + 1 observer” plan with resource numbers. Reference: https://prefecthq.github.io/prefect-kubernetes

Great question — by default, Prefect’s flow process does retry a bunch of transient failures, but not HTTP 500 specifically. That’s why you’re seeing pods exit when the API returns a 500. What Prefect retries automatically - Network/transport errors: timeouts, connect/read/write errors, connection resets, protocol errors - HTTP status codes: 408, 429, 502, 503 (and a CSRF corner of 403) - Exponential backoff with jitter, up to PREFECT_CLIENT_MAX_RETRIES (default 5) What it does not retry by default - HTTP 500. It’s treated as a non-idempotent server error and raised immediately. How to make your pods retry 500s instead of crashing - Set these env vars for your flow pods (work pool template env defaults are ideal):

PREFECT_CLIENT_RETRY_EXTRA_CODES=500
PREFECT_CLIENT_MAX_RETRIES=10
PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.5

- This augments the client’s built-in retry list to include 500 and increases attempts. All client calls made by the flow (e.g., read_flow_run, set_flow_run_state) will now retry on 500 with exponential backoff. Optional: increase client timeouts if you suspect slow API under load - You can pass httpx timeout settings when you create a client in your own code, but most users just rely on the defaults. If you have custom bootstrap code that calls the API early, wrap it with retries (or rely on the global env above). Why 500 isn’t retried by default - Some API operations aren’t perfectly idempotent (e.g., certain state changes). Prefect errs on the side of not repeating them unless you opt-in. If your flows are safe to retry on 500, adding it to

PREFECT_CLIENT_RETRY_EXTRA_CODES

is the sanctioned way to do it. Combine with the observer/API load fixes - Enabling 500 retries in pods will help them ride out transient server hiccups. - To reduce the number of hiccups in the first place, also implement: - Single, scoped observer; disable others - Disable pod-event replication - Gentle startup concurrency - Keep k8s authoritative on hard stops (backoffLimit: 0, ADS, restartPolicy: Never) If you’d like, I can produce a ready-to-paste patch for your work pool template env defaults that includes both: -

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=crash

and - the

PREFECT_CLIENT_RETRY_*

settings above.

These control the SQLAlchemy connection pool in the Prefect API process. They matter, but they’re a different layer than the observer/worker issues we’ve been discussing. What they do - PREFECT_SQLALCHEMY_POOL_SIZE - Baseline number of persistent DB connections per API process. - Default is 5. With 8 API replicas, that’s 8 × 5 = 40 baseline connections (plus services). - PREFECT_SQLALCHEMY_MAX_OVERFLOW - How many extra connections a process can open beyond pool_size under load. - Default is 10. With 8 replicas, worst case is 8 × (5 + 10) = 120 connections from the API alone. When to increase vs. when not to - Increase if: - You see “QueuePool limit reached”/pool timeout errors in API logs, or long waits to acquire a DB connection - Postgres has headroom (max_connections high enough; memory available) - Do not increase (or even reduce) if: - Postgres is near its connection limit or you see “too many connections” - You are not seeing pool exhaustion; raising these won’t help 500s caused by API saturation elsewhere At your scale (8 API replicas, 1000 concurrent jobs) - First, reduce API pressure from the observer side (single, scoped observer; disable pod-event replication). That will lower the 500s you’re seeing inside pods far more than DB pool tweaks. - Then, validate DB headroom and tune conservatively: - If you need more throughput and Postgres can handle it: - pool_size: 5–8 - max_overflow: 10–15 - If you need to cap DB connections tightly: - pool_size: 3–5 - max_overflow: 5–10 - Always calculate total possible connections: - Total ≈ (API replicas × (pool_size + max_overflow)) + services - Ensure this stays well under Postgres max_connections with buffer. Related settings worth checking - PREFECT_SERVER_SERVICES_EVENT_PERSISTER_BATCH_SIZE=20 (try 50–100 to reduce DB churn if event writes are heavy) - PREFECT_SERVER_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=5 - PREFECT_API_DATABASE_CONNECTION_TIMEOUT=5–10 (initial connection) - PREFECT_SERVER_DATABASE_SQLALCHEMY_POOL_TIMEOUT=30–60 (wait for pooled connection) If you’re heading toward high connection counts, strongly consider PgBouncer - Route the API to Postgres via PgBouncer in transaction mode and disable SQLAlchemy pooling (pool_size=null, prepared statement caches disabled). This lets you scale API replicas without multiplying persistent DB connections. Bottom line - These pool knobs are helpful, but they won’t fix 500s from API overload due to too many observers or high event volume. Implement the observer scoping/replication changes first; then adjust pool sizes if you see connection pool exhaustion in the API logs.

Short answer: yes, you’ll still see the important states. Disabling pod-event replication and scoping the observer only removes the chatty, pod-phase updates — it does not remove core visibility. What you will still see - Scheduled: The run is created/scheduled by Prefect (unaffected by observer settings). - Pending/Submitting: When a worker claims/submits the run (unaffected). - Running: When the flow code actually starts inside the pod, the flow engine sets the state to Running (unaffected). - Terminal states: Completed/Failed/Crashed/Cancelled will still be proposed by the observer (Job-level detection) or by the flow engine if it got that far. What you won’t see (by setting REPLICATE_POD_EVENTS=false) - Pod-phase-driven “infrastructure” states like InfrastructurePending when the pod is stuck in Pending or pulling an image. - Per-pod event spam (Pending → Running → Restarted, etc.) and automations driven by those pod events. Edge cases to be aware of - If the pod never starts (image pull error, scheduling failure), the flow engine will never set Running — but the Job will fail and the observer will still mark the Prefect run Crashed from the Job outcome. - If the observer cannot reach the API exactly when the Job fails, you can still miss the terminal proposal. That’s why I also recommend running a single, scoped observer and optionally adding a small reconciler CronJob for eventual correctness. Net effect - You keep clear visibility of “Scheduled → Pending → Running → Done/Failed/Cancelled”. - You trade away the granular pod-phase breadcrumbs to dramatically reduce API/event load at your scale. Docs for reference: prefect-kubernetes docs

Short answer: yes—moving from app-side polling to Prefect-triggered chaining will drastically reduce API calls. Given you’re on Prefect 2.x today, here are your best options to avoid (or nearly eliminate) polling: - Prefect Cloud Automations (recommended): Trigger the next deployment when the first flow run completes via the server-side event system. No client/app polling. The server listens for the

prefect.flow-run.Completed

event and issues a single RunDeployment action. This typically cuts calls from “every 5s for hours” down to “one call when it’s time to run.” - OSS 2.x with experimental events: You can enable events/automations server-side (no client polling) by setting: -

PREFECT_EXPERIMENTAL_EVENTS=true

-

PREFECT_API_SERVICES_TRIGGERS_ENABLED=true

- Subflows (if both jobs can live in the same codebase/process): Call the second flow as a subflow from the first. No polling and minimal API chatter, but the flows run under one parent flow and share the same runtime context. -

run_deployment

from a flow (still polls): If you call

prefect.deployments.run_deployment()

inside a flow and wait for completion, it polls the Prefect API every 5 seconds by default. This removes your app’s polling but not the polling overall. If you set

timeout=0

, it becomes fire-and-forget (no waiting), but then you still need an event/automation to start the next job at the right time. Key details for Prefect 2.x `run_deployment`: - It can wait for completion and returns a

FlowRun

with final state. - Default behavior is to poll every 5 seconds until the run is finished (or until

timeout

elapses). - You can adjust

poll_interval

or set

timeout=0

to not wait. Example:

from prefect import flow
from prefect.deployments import run_deployment

@flow
def orchestrator():
    # This will poll until the child run finishes (5s interval by default)
    fr = run_deployment(
        name="proj/first-job",
        parameters={"x": 1},
        timeout=None,         # wait indefinitely
        poll_interval=5       # default
    )
    # Then kick off the next job
    if fr.state.is_completed():
        run_deployment(name="proj/second-job", parameters={"y": 2}, timeout=0)  # fire-and-forget

If your goal is simply “submit job B as soon as job A completes” with minimal API traffic: - Use an Automation: trigger on

prefect.flow-run.Completed

for deployment A and action “Run deployment B.” This is server-side and event-driven—no polling loop. - If you’re strictly OSS and can’t enable events, consider converting to a single parent flow that calls both jobs as subflows. Want me to share exact, version-matching docs/snippets for your setup? If you can confirm: - Are you on Prefect Cloud or self-hosted OSS? - Your exact Prefect 2.x version? I’ll send precise links and a minimal Automation or subflow template tailored to that environment.

Here’s the recap we used for Prefect 2.x: add a retry_condition_fn that only returns True for 5xx responses, and raise an HTTP error from your request so Prefect can inspect it from the failed state. Example with requests:

import requests
from prefect import flow, task
from prefect.tasks import exponential_backoff  # optional

def retry_on_5xx(task, task_run, state) -> bool:
    try:
        # Will re-raise the exception that caused the task to fail
        state.result()
    except requests.HTTPError as exc:
        status = exc.response.status_code if exc.response is not None else None
        return status is not None and 500 <= status < 600
    except Exception:
        # Don’t retry other exception types (adjust if you want)
        return False
    return False  # No exception => no retry

@task(
    retries=3,
    retry_delay_seconds=5,  # or exponential_backoff(2) for 2s, 4s, 8s, ...
    retry_condition_fn=retry_on_5xx,
)
def fetch(url: str):
    r = requests.get(url, timeout=10)
    r.raise_for_status()  # raises requests.HTTPError on non-2xx
    return r.json()

@flow
def run():
    return fetch("<https://httpbin.org/status/500>")

If you’re using httpx instead of requests, swap the exception to

httpx.HTTPStatusError

and keep the same pattern. Want me to tailor this to include 429 (rate limiting) or use exponential backoff by default?

You’re right — in Prefect 3.x you can extend the client’s retryable HTTP status codes via an env var. - Env var:

PREFECT_CLIENT_RETRY_EXTRA_CODES

- What it does: Adds to Prefect’s built-in set of transient codes the client will retry on. Use it if you want to include 500 (or others like 504) as retriable. - Format: Comma-separated list (string) or a set when used via settings in code. Examples - Shell:

export PREFECT_CLIENT_RETRY_EXTRA_CODES="500"
# or multiple
export PREFECT_CLIENT_RETRY_EXTRA_CODES="500,504"

- PowerShell:

$env:PREFECT_CLIENT_RETRY_EXTRA_CODES = "500,504"

- In code:

from prefect.settings import temporary_settings, PREFECT_CLIENT_RETRY_EXTRA_CODES

with temporary_settings({PREFECT_CLIENT_RETRY_EXTRA_CODES: "500,504"}):
    # Prefect client will now retry on 500 and 504 in addition to its defaults
    ...

Related knobs -

PREFECT_CLIENT_MAX_RETRIES

— total retry attempts (default applies if unset) -

PREFECT_CLIENT_RETRY_JITTER_FACTOR

— adds jitter to backoff to avoid thundering herd Notes - This controls retries for Prefect’s own API calls. If you also need your task’s outbound HTTP calls (e.g., requests/httpx) to retry on 500, keep using task-level retries or a

retry_condition_fn

for that logic. - Be mindful of idempotency: some routes aren’t safe to retry.

You’re right—and sorry for the confusion earlier. Since you’re on Prefect 3, yes: moving from your app’s polling loop to Prefect’s event-driven chaining will greatly reduce API calls. What changes the API call profile in 3.x: - Automations and Deployment Triggers are server-side and event-driven. When flow A finishes, the server emits a

prefect.flow-run.Completed

event and immediately triggers flow B—no client-side polling. This typically cuts calls from “every 5s for the entire runtime” down to just the single action that starts the next deployment. -

run_deployment

still polls if you wait for completion. In 3.x it supports

timeout

(default: wait indefinitely) and

poll_interval

(default 5s). If you want to avoid polling from your orchestrator, set

timeout=0

to fire-and-forget and let an automation handle the chaining. Recommended 3.x patterns (to replace your app polling): 1) Automation: trigger B when A completes (no client polling)

from datetime import timedelta
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import RunDeployment

Automation(
    name="A-completed-triggers-B",
    trigger=EventTrigger(
        expect={"prefect.flow-run.Completed"},
        match_related={"prefect.resource.name": "A"},  # upstream deployment name
        posture="Reactive",
    ),
    actions=[
        RunDeployment(
            deployment_id="<downstream-deployment-id>",
            parameters={"upstream_run_id": "{{ flow_run.id }}"}
        )
    ],
).create()

- Docs: Automations concept, Chaining deployments with events, Creating automations 2) Deployment-level trigger (define on B to react to A)

from prefect import flow, serve
from prefect.events import DeploymentEventTrigger

@flow
def flow_a(): ...
@flow
def flow_b(): ...

a = flow_a.to_deployment(name="A")
b = flow_b.to_deployment(
    name="B",
    triggers=[
        DeploymentEventTrigger(
            expect={"prefect.flow-run.Completed"},
            match_related={"prefect.resource.name": "A"},
        )
    ],
)

if __name__ == "__main__":
    serve(a, b)

- Docs: Creating deployment triggers 3) If you must drive it from code:

run_deployment

trade-offs - Waiting = polling every

poll_interval

seconds (default 5s) - Avoid waiting/polling with

timeout=0

(submit and return immediately)

from prefect.deployments import run_deployment

# Non-blocking submit of B
run_deployment("B", parameters={"x": 1}, timeout=0)

# If you must wait (will poll):
run_deployment("A", timeout=300, poll_interval=10)

- Docs: Run deployments Why this reduces calls: - Your previous approach: 1 API call every 5s per running job until completion. - Event-driven approach: 0 polling calls; one action when the event fires. If you share whether you’re on Cloud or Server and the exact A→B flow names, I can give you a drop-in automation or trigger config tailored to your setup.

Short answer: it depends on which concurrency mechanism you mean. - Tag-based concurrency limits (Prefect 2.x): These apply to tasks, not flows. Flow runs do not consume tag-based concurrency slots at all. For tasks, a slot is held while the task is Running or Cancelling and is released as soon as it reaches a terminal state (Completed/Failed/Cancelled/Crashed). - Work queue/pool concurrency (flow-level caps): These limit how many flow runs can run concurrently from a given queue/pool. A flow that is Cancelling still counts against that cap until it reaches a final state (e.g., Cancelled). Once the flow actually transitions to a terminal state, the slot is freed and another run can start. Can you confirm whether you’re asking about: - task tag-based concurrency limits, or - work queue/pool flow-level concurrency caps?

Great question. Here’s what happens in Prefect 3.x with a “fire-and-forget” submission of 14,000 runs to a work pool capped at 1,000 concurrent flows: What happens when you submit 14,000 at once - The worker/pool model is pull-based. Workers fetch runs up to the available pool slots. - Only runs in PENDING or RUNNING count toward the work pool concurrency limit. - The first ~1,000 runs will move to PENDING and then RUNNING (as workers pick them up). - The remaining ~13,000 runs will sit in SCHEDULED (and may become Late if their scheduled time passes), but they do not consume pool concurrency slots until a worker pulls them. - As slots free up, the next batch of runs transitions to PENDING and then RUNNING, generally FIFO by scheduled time. Impact and load considerations - API/storage load: creating 14k runs is fine, but it’s a lot of rows and events at once; expect a brief spike in DB writes and event traffic at submission time. - Worker polling: workers poll periodically (default ~15s). They will only request up to the available slots per poll, and the API caps how many runs are returned per request. Queuing is handled server-side. - Late state: SCHEDULED runs that have passed their scheduled time will transition to “Late” after a threshold. “Late” is informational; they still don’t consume slots until picked up. If the parent is cancelled - Cancelling the parent flow does not automatically cancel all already-created child/sibling runs that are SCHEDULED in the pool. - In-process subflows created within the same engine context can cascade-cancel, but separately scheduled flow runs (what you’re doing here) will persist unless you cancel them explicitly. How to cancel the queued/scheduled jobs - Individually: -

prefect flow-run cancel <id>

- In bulk (recommended here): - Filter by deployment, tags, or other criteria and use the client or REST API to set their state to Cancelling/Cancelled. - Example (Python):

from prefect import get_client
    from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStateType, FlowRunFilterDeploymentId
    from prefect.states import Cancelling

    async def bulk_cancel_scheduled(deployment_id):
        async with get_client() as client:
            await client.set_flow_run_states(
                filters=FlowRunFilter(
                    deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]),
                    state_type=FlowRunFilterStateType(any_=["SCHEDULED", "LATE"]),
                ),
                state=Cancelling(),
                limit=10000,  # adjust and repeat if needed
            )

- You can also filter on creation time, tags applied at submission, or custom labels to target just this wave of 14k runs. Key takeaways - Submitting 14k runs to a pool with concurrency=1000 is supported: only ~1000 will be active; the rest wait in SCHEDULED/Late. - Parent cancellation does not auto-cancel already-submitted scheduled runs—you’ll need to bulk-cancel them. - For easier reversibility, consider tagging this wave of runs (e.g.,

batch:2026-04-15

) so you can filter and cancel/update them as a group later. If you share how you’re submitting those 14k (deployment name, any tagging you use), I can give you a ready-to-run bulk-cancel snippet tailored to your filters.

I get the frustration. Short answer: no—Prefect 3 can scale very well, but whether it “feels” fast or slow depends a lot on how the workload is modeled and how the control plane/workers are sized and tuned. Where Prefect 3 shines at scale - High run volumes with event-driven chaining: use Automations/Deployment Triggers (no polling overhead, low control-plane churn). - Many independent jobs: model them as separate flow runs (deployments) and scale workers/pool concurrency horizontally. - Dynamic, code-first orchestration with rich retries/timeouts and low boilerplate. Where you may need to be deliberate - Millions of very short-lived “tasks”: Prefect runs tasks inside a flow-run process by default; if you need cluster-wide parallelism per task, pair with a distributed task runner (Dask/Ray) or split work into many flow runs so the cluster can fan out. - Control plane/database sizing: bursts like 10–50k submissions create short-lived spikes in DB writes and events. Right-size Postgres (or use Cloud), tune worker poll intervals, and avoid app-side polling. Common bottlenecks (and fixes) - App-side polling of runs: switch to Automations/Triggers to eliminate N×5s API checks. - Under-provisioned DB/control plane: scale Postgres CPU/IO and API replicas; reduce log/artifact verbosity if it’s hot. - Worker underutilization: increase worker count, concurrency, and pool limits; ensure workers are close to the execution environment. - Overly chatty flows: minimize excessive logging/artifacts/results for very high-throughput runs. - Single giant flow with huge fan-out: use distributed task runners or split into subflows/deployments to spread across machines. When Airflow may look “faster” - If your workload is primarily “lots of tiny, independent tasks” and you’re using Celery/KubernetesExecutor per-task scaling, Airflow’s per-task worker model can feel snappier out of the box. In Prefect, you get equivalent parallelism by using distributed task runners or by modeling each unit of work as a flow run (so the pool/worker layer can scale it). If you’re seeing poor performance today, I’m happy to help tune it. The quickest way to diagnose is: - Cloud or self-hosted? - Avg run duration and daily run volume - Worker type/count/concurrency and poll interval - DB size/type (if self-hosted) - How you’re parallelizing (single flow with many tasks vs many flow runs; any Dask/Ray) - Logging/artifact/result settings Share a bit of that and I’ll suggest concrete, low-risk changes to speed things up.

Thanks for the details — that’s a serious workload, but it’s within range for Prefect 3 with the right tuning. Quick sizing sanity check - 14k–24k runs/day, each ~25 minutes: - Total minutes/day ≈ 350k–600k - Sustained parallelism needed ≈ 350k–600k / 1,440 ≈ 240–420 concurrent runs - A work pool concurrency of 1,000 is plenty; the key is to avoid control-plane spikes and keep the DB happy. Key recommendations for this pattern - Submit without waiting/polling: - Use

run_deployment(..., timeout=0, as_subflow=False)

to fire-and-forget and avoid creating a parent subflow task-run for each child (cuts extra records and orchestration churn). - Docs: Run deployments - Throttle or schedule in waves to prevent a submission spike: - Don’t push 14–24k creates in one second. Stagger them (e.g., 500–1,000 every 10–30s) and/or set

scheduled_time

to spread across a few minutes. This flattens DB write pressure and late-run noise. - Example:

from datetime import datetime, timedelta
    from prefect.deployments import run_deployment

    base = datetime.utcnow()
    batch = 500   # runs per wave
    cadence = 10  # seconds between waves

    for i, params in enumerate(payloads):  # 14k–24k items
        offset = (i // batch) * cadence
        run_deployment(
            "my-flow/my-deployment",
            parameters=params,
            scheduled_time=base + timedelta(seconds=offset),
            timeout=0,
            as_subflow=False,
            work_queue_name=f"shard-{i % 4}",  # shard across queues if you create them
        )

- Shard across multiple work queues in the same pool: - Create a few queues (e.g., 4–8) with their own workers. Sharding reduces lock contention and smooths fetches. Direct runs via

work_queue_name=...

. - Right-size workers to your sustained concurrency: - Target 300–450 active flow runs across your fleet (based on your volume/duration); keep pool limit a bit higher (e.g., 600–1,000) for headroom. Prefer many workers with moderate concurrency over a few huge ones. - Tame control-plane load: - Reduce logs on children (e.g., fewer prints,

log_prints=False

where applicable); keep log level at INFO or WARN. - Avoid unnecessary artifacts and result persistence for high-throughput children; only persist what you need. - Keep worker poll interval reasonable (10–15s is fine for 20–30 min runs). - If the parent is only there to fan out: - Consider replacing the 24h parent with an Automation or a small scheduled “submitter” flow that fans out in paced waves. Automations/Deployment Triggers are event-driven and keep API chatter low: - Automations concept - Chaining deployments with events - Creating deployment triggers Cancellation behavior (important here) - Cancelling the parent will not automatically cancel already-submitted SCHEDULED children in the pool. You’ll need to bulk-cancel them (e.g., filter by deployment, tags, or creation time) and set state to Cancelling/Cancelled. - Tip: tag each wave (e.g.,

batch:2026-04-15T10:00

) at submission so you can filter and bulk-cancel just that batch later. About your DB (RDS m6.2xlarge) - Likely OK if: - You throttle submissions (avoid one-shot 20k creates). - Autovacuum is healthy; storage has ample IOPS/throughput; and you scale API replicas if self-hosted. - You reduce log/artifact/result volume from children. - Watch for spikes in write IOPS and CPU during fan-out windows. If needed, bump IOPS or instance size, and consider sharding queues. If you can share: - Cloud or self-hosted? - Number of workers, per-worker concurrency, poll interval - Current logging/result settings for the children

- Whether you can shard into multiple queues …I’ll tailor exact queue/worker counts and a bulk-cancel snippet that matches your filters and infra.

Totally fair concern. “Fire-and-forget” doesn’t have to mean “we never know when they’re done.” In Prefect 3 you’ve got a few solid, low-churn ways to get a reliable “all done” signal without per-run polling. Pick the pattern that fits your needs: 1) Event-driven “batch complete” via Automations (no polling) - Tag every child run with a batch tag (e.g.,

batch:2026-04-15T10:00

). - Create a one-off Automation with an EventTrigger that: - expect:

prefect.flow-run.Completed

(or include Failed/Cancelled if you want “all finished, any outcome”) - match: scoped to that batch tag - threshold: the number of runs you submitted for that batch - within: a window long enough to cover the batch - Action: send a notification or trigger a downstream “completion” deployment. This fires exactly once when the Nth matching completion event arrives—no parent waiting, no API polling. Notes: - If you want “all finished regardless of outcome,” set expect to both Completed and Failed (and optionally Cancelled) and keep threshold = N. - Best practice: programmatically create this Automation right after you submit the batch (you know N there). Links: - Automations concept: https://docs.prefect.io/v3/concepts/automations/ - Chaining with events: https://docs.prefect.io/v3/how-to-guides/automations/chaining-deployments-with-events/ - Creating automations: https://docs.prefect.io/v3/how-to-guides/automations/creating-automations/ 2) Lightweight “batch aggregator” check (1 query every X minutes) - Tag runs at submission (e.g.,

tags=[batch_tag]

). - Periodically run a tiny aggregator flow that calls the Prefect API once to count runs by state for that tag and compares to N. This is a single batched API call per interval (e.g., every 1–5 minutes), not 14k polls. - When counts match your completion criterion (e.g., Completed == N or Final == N), trigger the next step and stop. Sketch:

# Pseudocode
batch_tag = "batch:2026-04-15T10:00"
# Submit children
for p in params_list:
    run_deployment("flow/dep", parameters=p, tags=[batch_tag], timeout=0, as_subflow=False)

# Later, a small aggregator flow:
async with get_client() as client:
    # Query counts for tag in a single call using filters
    # Compare completed_count (or final_count) to expected N
    # If done: notify/trigger downstream

3) Direct waiters (not recommended at this scale) -

wait_for_flow_run

exists, but waiting on 14k IDs just moves polling to a different place and increases churn. Key clarifications that change the recipe - Do you need “all succeeded” or “all finished (success or failure)”? Automations can watch only Completed, or Completed+Failed/Cancelled, and threshold accordingly. - Do you want a downstream flow to run at the end, or just a notification? That determines the Automation action (RunDeployment vs SendNotification). If you’d like, I can: - Show a concrete Automation snippet using a

batch_tag

with

threshold=N

for your batch size. - Share a minimal aggregator flow that does one filtered count query per interval and exits when done. Either way, you can keep

run_deployment(..., timeout=0, as_subflow=False)

for fan-out and still get a precise, low-overhead “all done” signal.

I hear you—that mismatch is maddening. If the pod is Error while the flow run still shows Running, something is preventing Prefect from learning the pod/job failed. In 3.x with the Kubernetes worker, the flow run should flip to Crashed within a few minutes when either: - the worker detects the Job/Pod failed, or - the engine heartbeats stop and the crash detector marks it Crashed. If it stays Running, the common culprits are: - Worker RBAC can’t read Jobs/Pods/Pod logs (so it never posts the failure back) - The pod crashed before the Prefect engine started (no heartbeats, plus worker couldn’t watch it) - The job/pod name/namespace doesn’t match what the worker is tracking (wrong namespace or template) - Missing/invalid PREFECT_API_* in the container, causing an immediate crash before state can update Let’s get precise signals: 1) Grab pod and container logs

# replace <ns> if not default
kubectl -n <ns> describe pod aboriginal-shrimp-wfh78-857ww
kubectl -n <ns> logs aboriginal-shrimp-wfh78-857ww --all-containers

2) Check the Job the worker created (labelled with the flow run) - In the Prefect UI, copy the flow run ID, then:

kubectl -n <ns> get job -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o yaml
kubectl -n <ns> get pods -l job-name=<JOB_NAME> -o wide

3) Share the worker logs around submission time - Look for lines like “Submitting flow run …”, “Created job …”, and any warnings/errors after. - If you can, start the worker with debug for more detail:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

(You don’t need to recreate the pool; just bump log level on the worker process.) 4) Verify the worker service account RBAC The worker’s service account needs at least get/list/watch on jobs, pods, and pods/log in the target namespace(s). If it can’t watch resources, Prefect won’t see the failure. - Minimal ClusterRole snippet to compare against:

apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
kind: ClusterRole
metadata:
  name: prefect-worker-minimal
rules:
- apiGroups: ["", "batch"]
  resources: ["pods", "pods/log", "jobs"]
  verbs: ["get", "list", "watch"]

Bind it (or a Namespaced Role) to the worker’s ServiceAccount. 5) Sanity-check the job template/env - Ensure the image has Prefect and your code/deps - API envs present:

PREFECT_API_URL

and (for Cloud)

PREFECT_API_KEY

- Correct namespace: the worker and jobs must agree on namespace (explicit in your work pool/job template) What to expect after fixes - On pod crash, the worker should post a Crashed state promptly; if the engine starts then dies, heartbeats should also mark Crashed within a few minutes. You should not see “Running forever” once RBAC and job wiring are correct. If you can share: - Prefect version - Worker logs (a few dozen lines around the submission) -

kubectl describe pod

+

kubectl logs

for that pod - Whether you’re using a custom job template and the namespace you target …I’ll pinpoint the cause and give you the exact change to make. Here’s the K8s worker guide for reference: Kubernetes worker how-to.

That stack trace shows the Prefect API itself returned a 500 (Internal Server Error) during a simple read_flow_run. That’s a server-side failure, not a client/network issue—so throwing more compute at the pod running your flow won’t fix it. We need to look at the API and database side. What typically causes 500s under heavy load - Database saturation: too many connections, lock contention, or statement timeouts. Symptoms in Postgres logs: “remaining connection slots are reserved”, “canceling statement due to statement timeout”, deadlocks, or slow I/O. - API pod resource pressure: CPU/memory starvation or too few replicas for the query volume. - Connection pool instability: frequent connect/disconnect churn to Postgres without pooling (or PgBouncer misconfiguration). - Long GC pauses or container OOMs that recover, leaving the client to see transient 500s. Immediate triage steps 1) Get API logs around the error time (they’ll include a Python traceback explaining the 500):

# Identify API pods
kubectl -n research-prefect3 get pods -l <http://app.kubernetes.io/name=prefect-server|app.kubernetes.io/name=prefect-server>

# Logs for the API container (adjust label/container name if different)
kubectl -n research-prefect3 logs <api-pod-name> --tail=500

2) Check Postgres for connection/timeout issues:

-- connections vs. limit
SELECT count(*) AS total_conns FROM pg_stat_activity;
SHOW max_connections;

-- look for timeouts/lock waits
SELECT pid, state, wait_event_type, wait_event, query
FROM pg_stat_activity
WHERE datname = current_database()
ORDER BY state, wait_event_type
LIMIT 50;

Also inspect your Postgres logs for timeouts, connection exhaustion, or I/O stalls during the same window. 3) Check API capacity - Number of API replicas and their CPU/memory requests/limits. - Any OOMKills or restarts:

kubectl -n research-prefect3 describe pod <api-pod-name>

4) Worker-side mitigation (while we fix server) - Add client retries for transient 500s (so brief API blips don’t crash runs):

# In your worker and job pods env
PREFECT_CLIENT_MAX_RETRIES=8
PREFECT_CLIENT_RETRY_EXTRA_CODES=500,504
# optional jitter to spread retries
PREFECT_CLIENT_RETRY_JITTER_FACTOR=0.2

This won’t fix a persistent 500, but it smooths over short spikes. Scale and stability recommendations for your volume - Use a connection pooler (PgBouncer in transaction mode) in front of Postgres. It drastically reduces connection churn from many workers/pods. - Ensure Postgres has enough max_connections (or rely on PgBouncer so the DB can keep max_connections lower and stable), sufficient CPU/IOPS, and healthy autovacuum. - Run multiple API replicas behind a Service and consider an HPA on CPU/requests per second. - Stagger your 14k–24k submissions (waves or scheduled_time offsets) to flatten DB write spikes. - Keep worker poll intervals reasonable (10–15s) and shard across multiple work queues to reduce hot-spotting on the same rows. Why this can cause the “pod Error but flow shows Running” - If the API returns 500s when the worker or engine tries to report/refresh state, the flow may stay “Running” until a later heartbeat/crash detector pass succeeds. Fixing the API/DB errors restores timely state transitions. To move forward quickly, can you share: - Prefect version, how the API is deployed (Helm chart values or manifests), API replica count/resources. - A few dozen lines of API pod logs (where the 500 occurred). - Any relevant Postgres log lines and current max_connections/pg_stat_activity snapshot. I’ll pinpoint the cause from those logs and give you concrete changes (e.g., PgBouncer config, API replica/limits, or DB settings) to stop the 500s.

Got it — two possibilities here, and the fix depends on which you mean: A) You cancel the parent flow run but it stays Running (never moves to Cancelling/Cancelled) - That’s a control-plane issue. Given you’ve seen server 500s, the state transition is likely failing. - What to do: - Check API pod logs at the time you click Cancel — there should be a traceback for the 500. - As a stopgap, force the state from a client to confirm the API can accept it:

from prefect import get_client
    from prefect.states import Cancelled

    async def force_cancel(flow_run_id: str):
        async with get_client() as client:
            await client.set_flow_run_state(flow_run_id, Cancelled())

- If this also throws, the API/DB is still unstable. We need the API logs + Postgres logs to resolve the 500s first; otherwise cancels can’t be recorded. B) You cancel the parent, but all the children (triggered via run_deployment) keep running - That’s expected: cancelling a parent does not automatically cancel separately submitted child flow runs (especially with

run_deployment(..., timeout=0, as_subflow=False)

). - You need to cancel the children explicitly. Easiest if you tagged the batch at submission. Bulk-cancel children (filter by tag or deployment)

from prefect import get_client
from prefect.client.schemas import FlowRunFilter, FlowRunFilterTags, FlowRunFilterStateType, FlowRunFilterDeploymentId
from prefect.states import Cancelling

async def bulk_cancel_by_tag(batch_tag: str, limit=10000):
    async with get_client() as client:
        await client.set_flow_run_states(
            filters=FlowRunFilter(
                tags=FlowRunFilterTags(all_=[batch_tag]),
                state_type=FlowRunFilterStateType(any_=["SCHEDULED","PENDING","RUNNING","LATE"]),
            ),
            state=Cancelling(),
            limit=limit,
        )

async def bulk_cancel_by_deployment(deployment_id: str, limit=10000):
    async with get_client() as client:
        await client.set_flow_run_states(
            filters=FlowRunFilter(
                deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]),
                state_type=FlowRunFilterStateType(any_=["SCHEDULED","PENDING","RUNNING","LATE"]),
            ),
            state=Cancelling(),
            limit=limit,
        )

Kubernetes-specific gotcha for actual pod teardown - Even if the server marks runs Cancelling, the Kubernetes worker must be able to delete the Job/Pod. Make sure the worker’s ServiceAccount has delete permissions:

apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
kind: ClusterRole
metadata:
  name: prefect-worker
rules:
- apiGroups: ["batch"]
  resources: ["jobs"]
  verbs: ["get","list","watch","create","patch","delete"]
- apiGroups: [""]
  resources: ["pods","pods/log"]
  verbs: ["get","list","watch","delete"]

- Also confirm the worker and jobs are using the same namespace and that the worker observer/watcher is running; on cancel you should see logs like “Received cancellation for flow run …, deleting Job …”. Next best steps - Tell me which case you’re hitting (A: parent won’t cancel at all, or B: children keep running). - If A: please paste the API pod traceback around the Cancel attempt; we’ll fix the 500 so state changes stick. - If B: confirm how you can filter the wave (tag, deployment, time window) and I’ll tailor the bulk-cancel snippet exactly to your filters. Reference: Kubernetes worker guide https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/

Understood. If “Cancel” isn’t taking effect, there are a few failure points to check. Given you’ve stabilized the API/DB and pods are running, the most common causes now are wiring/visibility between Prefect and your K8s jobs. Work through this checklist (in order): 1) Verify the server actually accepted the cancel - Does the flow run state change to Cancelling in the UI/API? - Quick check:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<FLOW_RUN_ID>")
    print(fr.state_name, fr.state_type)  # Expect: Cancelling/Cancelled soon after clicking Cancel

- If this still shows Running after you click Cancel, we need the API logs at cancel time; the server may be rejecting the transition silently (rare) or a policy is blocking it. 2) Confirm you’re cancelling the right thing - Cancelling a “parent” does not cancel independently submitted child runs created via

run_deployment(..., as_subflow=False)

. If your goal is to stop the entire batch, bulk-cancel the children (filter by deployment/tag/time window). I can tailor a snippet if you tell me which filter you can use. 3) Ensure the worker can find and delete the Job/Pod for a cancelled run - Required labels/namespace must be preserved in your custom job template. If the worker cannot locate the job by label, it cannot cancel it. - Your Job and Pod must include:

metadata:
      labels:
        <http://prefect.io/flow-run-id|prefect.io/flow-run-id>: "{{ flow_run.id }}"
    spec:
      template:
        metadata:
          labels:
            <http://prefect.io/flow-run-id|prefect.io/flow-run-id>: "{{ flow_run.id }}"

- Namespace: confirm the job runs in the same namespace the worker expects (via your pool/job variables). Mismatched namespaces = worker can’t see/delete it. - Sanity check from the cluster:

kubectl -n <ns> get job -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>>
kubectl -n <ns> get pods -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o wide

If these return nothing, the labels are missing and the worker cannot act on cancellation. 4) Verify worker RBAC includes deletes - Without delete, the worker sees the cancel but cannot tear down the job/pod. - Minimal example:

apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
    kind: ClusterRole
    metadata:
      name: prefect-worker
    rules:
    - apiGroups: ["batch"]
      resources: ["jobs"]
      verbs: ["get","list","watch","create","patch","delete"]
    - apiGroups: [""]
      resources: ["pods","pods/log"]
      verbs: ["get","list","watch","delete"]

- Bind to the ServiceAccount your worker uses; ensure it’s valid in the job’s namespace(s). 5) Watch the worker react to your cancel - Increase worker logging and cancel a test run:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

- You should see logs like “observed cancellation for flow run …” followed by “deleting Kubernetes Job …”. If you see neither, the worker isn’t observing state changes for runs it launched (often due to missing labels or wrong namespace). If you see errors on delete, that’s RBAC. 6) Rule out customizations by isolating - Spin up a stock worker (no custom job template) in a test pool and run/cancel a hello-world flow. If cancel works there, the issue is in your custom job template, namespace, or RBAC. If it still doesn’t, share the debug logs; we’ll look deeper. Optional, to immediately stop a batch of children - If the parent’s cancel can’t cascade (expected for independently submitted runs), bulk-cancel the children by tag/deployment: ``` from prefect import get_client from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterTags, FlowRunFilterDeploymentId, FlowRunFilterStateType from prefect.states import Cancelling async def bulk_cancel_by_tag(tag, limit=10000): async with get_client() as c: await c.set_flow_run_states( filters=FlowRunFilter( tags=FlowRunFilterTags(all_=[tag]),

state_type=FlowRunFilterStateType(any_=["SCHEDULED","PENDING","RUNNING","LATE"]), ), state=Cancelling(), limit=limit, ) ``` Most frequent root causes I see in your situation - Missing

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

label in a customized job template → worker can’t find the job to delete. - RBAC lacks delete on jobs/pods or wrong namespace → delete fails silently in worker logs unless DEBUG. - Expecting parent cancel to cascade to separately submitted children → this is not automatic. If you can provide: - Output of those kubectl label queries for an affected run - Your job template metadata (labels + namespace) - A short DEBUG excerpt from the worker after you click Cancel …I’ll pinpoint the exact gap and give you the one-line fix. For reference: https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/

Short answer: yes—your Job and Pod need the Prefect labels, especially

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

. The Kubernetes worker relies on those labels to find and delete the right Job/Pod on cancel. Right now your template only applies

{{ labels }}

at the Job metadata level, and the Pod template metadata has a hardcoded

app

label but does not include the Prefect labels. Add the labels to both places. Minimal fix (showing the important bits):

"job_manifest": {
  "apiVersion": "batch/v1",
  "kind": "Job",
  "metadata": {
    "namespace": "{{ namespace }}",
    "generateName": "{{ name }}-",
    "labels": {
      "app": "prefect3-DNR",
      "<http://prefect.io/flow-run-id|prefect.io/flow-run-id>": "{{ flow_run.id }}",
      "<http://prefect.io/deployment-id|prefect.io/deployment-id>": "{{ deployment.id }}",
      "<http://prefect.io/work-pool-id|prefect.io/work-pool-id>": "{{ work_pool.id }}"
      // include any of your own labels too, or just merge {{ labels }} here
    }
  },
  "spec": {
    "template": {
      "metadata": {
        "labels": {
          "app": "prefect3-DNR",
          "<http://prefect.io/flow-run-id|prefect.io/flow-run-id>": "{{ flow_run.id }}",
          "<http://prefect.io/deployment-id|prefect.io/deployment-id>": "{{ deployment.id }}",
          "<http://prefect.io/work-pool-id|prefect.io/work-pool-id>": "{{ work_pool.id }}"
          // IMPORTANT: include the same labels on the Pod template
          // If you prefer, you can use {{ labels }} here to include Prefect + your labels
        },
        "annotations": {
          "<http://karpenter.sh/do-not-disrupt|karpenter.sh/do-not-disrupt>": "true",
          "<http://cluster-autoscaler.kubernetes.io/safe-to-evict|cluster-autoscaler.kubernetes.io/safe-to-evict>": "false"
        }
      },
      "spec": {
        "serviceAccountName": "{{ service_account_name }}",
        ...
      }
    },
    "backoffLimit": 0,
    "ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
  }
}

Notes: - You can also just inject

{{ labels }}

into both

metadata.labels

and

spec.template.metadata.labels

if your worker’s

labels

variable already includes the Prefect system labels; the key is that the Pod template must carry them as well. - Be careful with Slack’s rendering: keys like

<http://nvidia.com/gpu|nvidia.com/gpu>

should be literally

<http://nvidia.com/gpu|nvidia.com/gpu>

in your manifest (not a hyperlink). Same for

<http://topology.kubernetes.io/zone|topology.kubernetes.io/zone>

. Validate quickly:

# After launching a flow run, verify both the Job and Pod have the flow-run label
kubectl -n <ns> get job -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o name
kubectl -n <ns> get pod -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o name

Both commands should return a resource name. If the Pod query returns nothing, the worker can’t locate/delete it on cancel. Also ensure RBAC allows deletes (for actual teardown):

- apiGroups: ["batch"]
  resources: ["jobs"]
  verbs: ["get","list","watch","create","patch","delete"]
- apiGroups: [""]
  resources: ["pods","pods/log"]
  verbs: ["get","list","watch","delete"]

If you add the labels to the Pod template and have delete RBAC, you should see the worker log “deleting Kubernetes Job …” on cancel and the run move from Cancelling to Cancelled. If it still doesn’t, please share: - The rendered Job YAML for an affected run (metadata + template.metadata labels) - The worker logs (DEBUG) right after you click Cancel Reference: https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/

Thanks for sharing the template. Two parts to answer your question clearly: 1) Do I need to hardcode prefect.io/* labels in my template? - If you include

{{ labels }}

in your Job metadata AND in the Pod template metadata, the Kubernetes worker will populate that dict with Prefect’s system labels at runtime and merge in your custom labels. You do not need to manually write: -

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

-

<http://prefect.io/deployment-id|prefect.io/deployment-id>

-

<http://prefect.io/work-pool-id|prefect.io/work-pool-id>

- Those will be injected automatically as long as: - The worker is rendering your template (i.e., using a Kubernetes work pool/worker) -

labels

is applied in both Job metadata and spec.template.metadata - In your snippet, I see

metadata.labels: "{{ labels }}"

on the Job, but your Pod template metadata has its own

labels

block that does not include

{{ labels }}

. Add

{{ labels }}

there too so the Prefect labels are present on both the Job and the Pod. Minimal change to your template - Add the

{{ labels }}

to the Pod template:

"spec": {
  "template": {
    "metadata": {
      "labels": {
        "app": "prefect3-DNR",
        {{ labels }}
      },
      "annotations": {
        "<http://karpenter.sh/do-not-disrupt|karpenter.sh/do-not-disrupt>": "true",
        "<http://cluster-autoscaler.kubernetes.io/safe-to-evict|cluster-autoscaler.kubernetes.io/safe-to-evict>": "false"
      }
    },
    "spec": {
      ...
    }
  },
  "backoffLimit": 0,
  "ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
}

- Or, if you prefer explicit keys, you can hardcode:

"<http://prefect.io/flow-run-id|prefect.io/flow-run-id>": "{{ flow_run.id }}",
"<http://prefect.io/deployment-id|prefect.io/deployment-id>": "{{ deployment.id }}",
"<http://prefect.io/work-pool-id|prefect.io/work-pool-id>": "{{ work_pool.id }}"

But this is not required if you propagate

{{ labels }}

to the Pod template. 2) Shouldn’t the worker already have delete permissions? - If you deployed the Kubernetes worker with the official Prefect Helm chart and did not change the ServiceAccount/Role/RoleBinding, yes—the default RBAC includes delete on Jobs and get/list/watch on Pods/Pods logs in the worker’s namespace. - If you customized the ServiceAccount or RBAC (e.g., different SA, restricted Role, different namespace), it’s possible delete is missing. That would prevent the worker from tearing down Jobs/Pods on cancel. - Quick verification: - Identify the ServiceAccount your worker uses (value of

serviceAccountName

on the worker Deployment). - Check its effective verbs: - Jobs: get, list, watch, create, patch, delete - Pods: get, list, watch, delete; pods/log: get - You can test quickly by trying to delete a test Job with that SA (impersonation) or checking worker logs on cancel at DEBUG—they’ll show a forbidden error if RBAC is the issue. Actionable next steps - Propagate

{{ labels }}

into

spec.template.metadata.labels

in your job template. - Verify the labels appear on both the Job and the Pod for a new run: -

kubectl -n <ns> get job -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>>

-

kubectl -n <ns> get pod -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>>

- If cancels still don’t work, enable DEBUG on the worker and cancel a run; share the relevant log lines so we can see whether it’s a label lookup or RBAC issue. If you’d like, paste the rendered Job YAML for an affected run (just metadata sections), and I’ll confirm the labels are correct.

Great question, and thanks for sticking with the details. How cancellation works in the Prefect 3 Kubernetes worker - The worker does not search by labels to cancel. It uses the infrastructure_pid stored on the flow run, which is formatted as namespace:job_name. - On cancel, the worker calls the Kubernetes API to delete the Job directly: - delete_namespaced_job(name=<job_name>, namespace=<namespace>, grace_period_seconds=30, propagation_policy="Foreground") - Foreground means Kubernetes deletes the Job only after its owned Pods are terminated; pods get SIGTERM, then SIGKILL after the grace period. So do labels matter for cancellation? - Labels are not used to find the Job for deletion. The worker already knows the job_name and namespace from infrastructure_pid. - However, Prefect still injects labels on the Job and Pod for: - observability and troubleshooting (kubectl -l prefect.io/flow-run-id=…) - consistent metadata across resources - some auxiliary watchers/observers - Critically: you must ensure the Job name and namespace the worker created are the same ones encoded in the infrastructure_pid; if your template mutates these unexpectedly, deletion will miss the target. Why adding labels on the Pod (but not Job) didn’t fix cancel - Since the worker deletes the Job, not the Pod directly, if your template produces a Job name/namespace that doesn’t match the infrastructure_pid, the worker’s delete call won’t hit anything. Labels won’t help in that case. - Also, if your RBAC lacks delete on Jobs (or you’re in the wrong namespace), the delete call will fail and the Job/Pod will keep running. What to verify now 1) Confirm the infrastructure_pid and the actual Job match: - In the UI/API, copy the flow run’s infrastructure_pid (should be namespace:jobname). - Then:

kubectl -n <namespace_from_pid> get job <jobname_from_pid> -o yaml

If not found, your template likely altered the name/namespace (e.g., generateName overrides, suffixing, or using a different namespace). 2) Check your template isn’t overriding the name - The worker will set job metadata.generateName or name; if your template sets its own name/generateName differently from the worker’s computed name, the final created resource may not match the stored job_name. - Keep the worker-generated name/generateName field intact, or ensure your template uses the value rendered by the worker variables. 3) Ensure RBAC can delete Jobs - The ServiceAccount used by the worker must have delete on jobs in the target namespace(s):

- apiGroups: ["batch"]
  resources: ["jobs"]
  verbs: ["get", "list", "watch", "create", "patch", "delete"]

- If you didn’t change SA/RBAC via Helm, you’re likely fine; if you customized, double-check. 4) Watch the worker when you cancel - Start the worker with DEBUG and cancel a run:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <pool>

You should see a line like “Deleting Kubernetes job 'X' in namespace 'Y'”. If you see a 404/403 or other ApiException, that’s the smoking gun. TL;DR - Cancellation is done by deleting the Job via namespace:job_name from infrastructure_pid; labels are not used for that lookup. - Labels are still good practice, but if cancel isn’t working, focus on: - Does infrastructure_pid match the actual Job name/namespace created? - Does the worker SA have delete on Jobs in that namespace? - Is your template changing the Job name/generateName or namespace away from what the worker records? If you paste one real infrastructure_pid and the corresponding kubectl get job output (metadata.name/namespace), I’ll tell you exactly what’s mismatched.

Short answer: yes — the Kubernetes worker automatically injects Prefect’s system labels onto BOTH the Job and the Pod template at creation time. You do not need to add them manually. Important gotchas that likely caused your errors: - Do not paste Slack-formatted keys. Keys like

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

are invalid label keys in Kubernetes. Use the literal keys: -

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

-

<http://prefect.io/flow-run-name|prefect.io/flow-run-name>

-

<http://prefect.io/deployment-id|prefect.io/deployment-id>

-

<http://prefect.io/deployment-name|prefect.io/deployment-name>

-

<http://prefect.io/work-pool-id|prefect.io/work-pool-id>

-

<http://prefect.io/work-pool-name|prefect.io/work-pool-name>

-

<http://prefect.io/flow-id|prefect.io/flow-id>

-

<http://prefect.io/flow-name|prefect.io/flow-name>

-

<http://prefect.io/version|prefect.io/version>

-

<http://prefect.io/worker-name|prefect.io/worker-name>

- You also don’t need to template

{{ flow_run.id }}

etc. The worker derives and injects these values on its own. If you supply

{{ labels }}

in your template, that’s only for adding your own custom labels; Prefect’s system labels are merged in automatically and propagated to the Pod template. How to verify the worker’s injection is working - Launch a run, copy its flow run ID, then:

kubectl -n <ns> get job -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o name
kubectl -n <ns> get pod -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o name

Both should return resources. If the Job shows up but the Pod does not, your template may be overwriting the Pod template labels instead of allowing the worker to merge them. Reminder on cancellation mechanics - Prefect cancels by deleting the Job directly using the stored infrastructure_pid (namespace:job_name), not by searching with labels. The labels are for observability and tooling, but cancellation depends on: - The recorded job name/namespace matching the actual Job that was created - The worker’s ServiceAccount having delete permissions on Jobs in that namespace - Quick RBAC check:

kubectl auth can-i delete jobs --as=system:serviceaccount:<ns>:<worker-sa> -n <target-ns>

If you’re still seeing cancels not take effect: - Confirm infra_pid matches a real Job:

kubectl -n <ns_from_pid> get job <job_from_pid> -o yaml

- Start the worker with DEBUG, click Cancel, and look for “Deleting Kubernetes job …” or any ApiException details:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

Bottom line: - Yes, the worker injects all those Prefect labels automatically on both Job and Pod. - Don’t add Slack-linked keys; use literal

<http://prefect.io/|prefect.io/>...

keys only if you’re adding your own labels. - If cancellation still fails, it’s almost always infra_pid mismatch or missing delete RBAC — we can pinpoint with the two kubectl checks above plus a short worker DEBUG log excerpt.

Thanks for checking RBAC — that’s good news. Since the worker SA can delete Jobs in that namespace, the next most likely reasons a cancel “does nothing” are: 1) The worker is not actually issuing the delete (observer not firing) 2) The delete is targeting the wrong Job (infra_pid mismatch) 3) The Job is in a different namespace than the worker expects Here’s a focused checklist to pinpoint which one it is: A) Verify the exact target the worker would delete - Get the infra_pid and current state for an affected run:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<FLOW_RUN_ID>")
    print(fr.infrastructure_pid, fr.state_name, fr.state_type)

infra_pid must be namespace:job_name. - Does that Job exist?

kubectl -n <namespace_from_pid> get job <job_from_pid> -o yaml

If this returns NotFound, the worker is trying to delete a name that doesn’t exist (usually due to template name/generateName overrides or a different namespace used at creation time). B) Confirm the run is in the same pool the worker is serving - In the UI or via the client, confirm the run’s work pool matches the pool passed to

prefect worker start --pool ...

. - If the run went to a different pool (or queue), your worker will not observe its cancellation. C) Watch the worker react to your cancel (this is the single most telling step) - Start the worker with debug and cancel a test run:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

- After you click Cancel, you should see logs like: - “Received cancellation for flow run …” - “Deleting Kubernetes job '<job_from_pid>' in namespace '<ns_from_pid>'” - Or an ApiException (404/403/etc.) If you see nothing, the FlowRunCancelling observer isn’t firing (worker not subscribed/running). If you see an error, that’s our root cause. D) Namespace sanity check - Your template sets namespace via

{{ namespace }}

. Ensure that’s the same namespace encoded in infra_pid and the same namespace your RBAC check covered. - It’s common to have the worker in namespace A, but jobs created in namespace B. That’s fine, as long as the worker SA has delete in B (you can verify with:

kubectl auth can-i delete jobs --as=system:serviceaccount:<worker-ns>:<worker-sa> -n <target-job-ns>

). E) Name overrides in the template - Prefer using only

metadata.generateName

(and let Kubernetes assign the final name). The worker records the final Job name returned by the API and stores it in infra_pid. - If the template sets a conflicting fixed

metadata.name

, the created Job may not match what the worker stored, and cancels will miss. - If you can, share the rendered Job metadata (name, generateName, namespace) for one stuck run alongside the infra_pid. About labels - Yes, the worker automatically injects the Prefect labels onto both Job and Pod. You don’t need to add them manually. Using Slack-linked keys (e.g.,

<http://prefect.io/flow-run-id|...>

) will cause manifest errors — the literal keys are

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

, etc. - Labels are useful for kubectl lookups but are not used for the actual delete; the worker deletes by infra_pid (namespace:job_name). Quick litmus tests - Manual delete using the exact infra_pid (to ensure K8s path is fine):

kubectl -n <ns_from_pid> delete job <job_from_pid> --grace-period=30 --cascade=foreground

If this works, K8s is OK; the issue is the worker not invoking delete (observer) or targeting the wrong name/namespace. If you can share: - One stuck run’s infra_pid and the output of

kubectl -n <ns> get job <name> -o yaml

- A small DEBUG excerpt from the worker immediately after you click Cancel …I’ll tell you exactly whether it’s an observer/subscription gap, a name/namespace mismatch, or something else.

Thanks for the clear signal. If there are zero “cancel” messages in the worker logs when you click Cancel, the worker’s cancellation observer is not firing. At that point, the issue is not RBAC or labels — it’s that the worker never received a cancellation signal to act on. Here’s the shortest path to isolate and fix it: 1) Confirm the server accepted your cancel - When you click Cancel, does the flow run’s state actually change to Cancelling in the UI/API? - Quick check:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<FLOW_RUN_ID>")
    print(fr.state_name, fr.state_type)  # Expect: Cancelling → Cancelled shortly after

- If it stays Running, the cancel isn’t being recorded server-side. We need API logs at the cancel time. 2) Verify the worker is running the cancellation observer - Start the worker with DEBUG and watch its startup:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

You should see logs indicating observers are starting, including something akin to “FlowRunCancellingObserver”. If you do not see that, your changes likely disabled or broke the observer path. Restarting with defaults or upgrading to the latest 3.x typically restores it. 3) Make sure the run belongs to the worker’s pool - The cancellation observer only processes cancellations for runs in the pool the worker is serving. - Check the run’s work_pool_name/work_queue_name:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<FLOW_RUN_ID>")
    print(fr.work_pool_name, fr.work_queue_name, fr.infrastructure_pid)

- If the pool doesn’t match the one your worker serves, that worker will never see the cancel. 4) Ensure infra_pid is present and points to a real Job - Prefect cancels by deleting the Job using infra_pid (namespace:job_name). - Sanity check:

# Using values printed above
kubectl -n <namespace_from_infra_pid> get job <job_from_infra_pid> -o name

If NotFound, the server/worker recorded a different name/namespace than was created (name override, different ns, etc.), so deletion won’t target the right job. 5) Restart the worker to reset observers - If the worker was running during prior API instability, its observer subscription can be stuck. A clean restart often reestablishes the cancellation stream. Keep it on DEBUG to confirm you see messages when cancelling a test run. Important clarifications - Yes, the Kubernetes worker automatically injects Prefect labels onto both the Job and the Pod at creation. You don’t need to add them manually. Using Slack-rendered keys (e.g., …) will error — the literal keys are

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

, etc. - Labels are not how cancellation is executed. The worker deletes the Job via infra_pid (namespace:job_name). Labels are for observability. - Your RBAC check shows you can delete jobs — good. If the observer fires, deletion should work unless infra_pid doesn’t match a real job. If you can share: - Whether the flow run actually transitions to Cancelling in the UI/API - The worker’s DEBUG startup lines (do you see a cancellation observer start?) - The run’s work_pool_name/work_queue_name/infra_pid and the result of

kubectl get job

for that name/namespace …I’ll pinpoint whether it’s a server-side cancel not recording, a pool mismatch, an observer disabled/stuck, or an infra_pid/name/namespace mismatch, and give you the exact change to make. For reference: https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/

Yes — that’s almost certainly it. In Prefect 3, the worker only reacts to flow-run cancellations if its cancellation observer is enabled. There are two paths: - Events subscriber (primary): listens for prefect.flow-run.Cancelling events - API polling (fallback): periodically polls for runs in Cancelling If you disabled observers or specifically turned off worker-side cancellation, the worker will never receive or act on cancel requests. Given you said there are no “cancel” lines in the worker logs, this matches exactly. What to set back - Re-enable the worker’s cancellation observer: - Env var:

PREFECT_WORKER_ENABLE_CANCELLATION=true

- Or in config:

[worker]
    enable_cancellation = true

- Optionally tighten polling (fallback when events aren’t available):

PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60   # default is ~120s

- Run the worker with DEBUG to verify it starts the observer and reacts when you click Cancel:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

How it works under the hood - With enable_cancellation=true, the worker creates a FlowRunCancellingObserver at startup. - It first tries to subscribe to the events stream; if that fails, it falls back to polling the API at the configured interval. - On receiving a cancel for a run in its pool, it calls kill_infrastructure using the run’s infrastructure_pid (namespace:job_name), issuing a Kubernetes delete Job with propagation_policy=Foreground and a grace period. Why your cancels were ignored - With the observer disabled, the server flips runs to “Cancelling,” but the worker never observes that transition, so it never deletes the Job/Pod. That’s why there were no cancel messages in logs and nothing happened cluster-side. Quick validation plan 1) Set

PREFECT_WORKER_ENABLE_CANCELLATION=true

on the worker deployment and restart. 2) Start the worker with DEBUG and confirm you see the cancellation observer start. 3) Cancel a test run; you should see logs indicating the cancel event/poll was received and that the Job is being deleted. 4) If you want the runner process to be stricter in your serve-based deployments, you can also set:

PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE=true

That makes cancellation observation failures crash the runner instead of silently continuing. If you restore that setting and still don’t see cancel logs, share the worker’s DEBUG startup lines and I’ll double-check that the observer is actually being created and bound to your pool.

Short answer: an RDS Proxy is very unlikely to be the reason your worker never sees cancellations. The worker learns about cancels from the Prefect API (events WebSocket with API fallback polling), not from Postgres directly. If there are zero “cancel” lines in worker logs, the worker’s cancellation observer isn’t running or isn’t pointed at the same pool/API — not a DB/proxy issue. Focus on these checks: 1) Verify the server is actually recording the cancel - After clicking Cancel, confirm the run flips to Cancelling in the API:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<FLOW_RUN_ID>")
    print(fr.state_name, fr.state_type)  # Expect: Cancelling → Cancelled

If it stays Running, this is a control-plane issue (check API logs at cancel time). If it shows Cancelling, proceed. 2) Ensure the worker’s cancellation observer is enabled - Inside the worker pod, check config:

prefect config view | grep -i cancellation
# Expect:
# worker.enable_cancellation = true
# worker.cancellation_poll_seconds = 120 (or your override)

If

worker.enable_cancellation

is false or missing, set:

PREFECT_WORKER_ENABLE_CANCELLATION=true
# Optional for faster fallback polling:
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60

Restart the worker and run it with DEBUG:

PREFECT_LOGGING_LEVEL=DEBUG prefect worker start --pool <your_pool>

You should see it start a FlowRunCancellingObserver and either “subscribing to events” or “falling back to polling”. 3) Confirm pool and API alignment - The run must belong to the same work pool your worker serves:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<FLOW_RUN_ID>")
    print(fr.work_pool_name, fr.work_queue_name, fr.infrastructure_pid)

- Ensure the worker’s

PREFECT_API_URL

(and API key if Cloud) matches the API you’re cancelling against. A mismatched API endpoint means the worker is watching a different control plane. 4) Sanity-check infra_pid and namespace - Prefect cancels by deleting the Job from

infrastructure_pid

(namespace:job_name). Make sure it points to a real Job:

kubectl -n <ns_from_pid> get job <job_from_pid> -o name

If NotFound, your template likely changed the final job name or namespace; the worker is trying to delete the wrong job. 5) Watch the cancel path live - With DEBUG on the worker, click Cancel for a test run. You should see: - “received cancellation for flow run …” - “deleting Kubernetes job '<name>' in namespace '<ns>'” If you see nothing, the observer isn’t running; if you see ApiException, share that error. About the RDS Proxy - If the API/DB were failing to record the cancel, you’d see the run not flip to Cancelling or API 5xx at cancel time. But “no cancel logs at all” in the worker points to the observer not running or not subscribed — not a DB proxy problem. - Even if events WebSocket were flaky, the worker falls back to API polling (if the observer is enabled), so you’d still see periodic cancel processing in logs. If you want, paste: - Output of the config view lines from the worker pod - The run’s

work_pool_name

,

work_queue_name

, and

infrastructure_pid

- A short DEBUG excerpt from the worker after a cancel attempt I’ll pinpoint whether it’s observer disabled, pool/API mismatch, or infra_pid/name/namespace mismatch. References: Kubernetes worker how-to, Settings reference.

Those env vars you enabled are for the optional Kubernetes Observer integration that mirrors K8s pod/job events into Prefect’s event stream. They do not control (or replace) the worker’s cancellation observer. To make the worker actually receive and act on flow-run cancellations, add these to the worker container: - PREFECT_WORKER_ENABLE_CANCELLATION=true - (optional) PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60 - PREFECT_LOGGING_LEVEL=DEBUG (temporarily, to confirm you see the observer start and react) Example (since you’re already using extraEnvVars):

extraEnvVars:
  - name: PREFECT_WORKER_ENABLE_CANCELLATION
    value: "true"
  - name: PREFECT_WORKER_CANCELLATION_POLL_SECONDS
    value: "60"
  - name: PREFECT_LOGGING_LEVEL
    value: "DEBUG"

Then restart the worker and validate: 1) Inside the worker pod, confirm config:

prefect config view | grep -i worker.enable_cancellation
# Expect: worker.enable_cancellation = true

2) Start/monitor the worker logs. On startup you should see it creating a FlowRunCancellingObserver (it’ll try events first, then fall back to polling if needed). 3) Trigger a run in the same work pool the worker serves, click Cancel, and look for logs like: - “received cancellation for flow run …” - “deleting Kubernetes job '<name>' in namespace '<ns>'” If you still see no cancel logs after this: - Double-check the run’s pool matches the worker’s pool (the observer only processes cancellations for its pool). - Verify the worker’s PREFECT_API_URL points to the same API you’re using to click Cancel. - Print the run’s infrastructure_pid and confirm it’s a real Job:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<RUN_ID>")
    print(fr.work_pool_name, fr.infrastructure_pid)  # ns:jobname

kubectl -n <ns_from_pid> get job <job_from_pid> -o name

Keep the K8s observer vars if you want K8s-to-Prefect event mirroring, but they’re orthogonal to cancellation. The key switch for cancel handling is PREFECT_WORKER_ENABLE_CANCELLATION on the worker.

Short answer: no — it’s off by default in Prefect 3. You need to opt in with

PREFECT_WORKER_ENABLE_CANCELLATION=true

for workers to react to server-side cancels. Why it’s not enabled by default: - Least-privilege/safety: many installs don’t grant delete on Jobs/Pods; auto-enabling could cause unexpected infra deletions or RBAC errors. - Back-compat: some users rely only on in-process (runner) cancellation, not worker-driven infra teardown. - Multi-tenant/cross-namespace setups where blanket deletion isn’t appropriate. What to do: - Set on the worker and restart:

PREFECT_WORKER_ENABLE_CANCELLATION=true
# optional fallback polling interval
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60
# temporarily increase logs to confirm observer starts
PREFECT_LOGGING_LEVEL=DEBUG

- Verify inside the pod:

prefect config view | grep -i worker.enable_cancellation

- Then cancel a test run and look for delete logs. References: - Kubernetes worker how-to - Settings reference

You can absolutely click Cancel in the dashboard — that always updates the flow run’s state on the server. What’s not “on by default” is the part that actually tears down external infrastructure (e.g., deleting a Kubernetes Job/Pod). For that, the worker needs to be told to act on cancels. So the default behavior is: - Dashboard Cancel → server marks the run Cancelling/Cancelled - Stopping the compute depends on how you’re running the flow - In‑process (runner/serve): the process observes cancel and exits - Worker‑launched infra (Kubernetes/Docker/etc.): the worker must observe the cancel and kill the Job/Pod In Prefect 3, worker‑side cancellation is opt‑in: - Set this on your worker and restart it:

PREFECT_WORKER_ENABLE_CANCELLATION=true
# optional: faster fallback polling if events aren’t available
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60
# temporarily: to confirm you see cancel handling in logs
PREFECT_LOGGING_LEVEL=DEBUG

Then validate: - Inside the worker pod:

prefect config view | grep -i worker.enable_cancellation

- Cancel a test run in the same work pool; you should see logs like: - “received cancellation for flow run …” - “deleting Kubernetes job '<name>' in namespace '<ns>'” Why it’s not enabled by default: - Safety/least‑privilege: many clusters don’t grant delete on Jobs/Pods by default; auto‑deletes could surprise users - Back‑compat and multi‑tenant concerns If you’ve enabled the var and still don’t see cancel logs: - Confirm the run’s work_pool_name matches the pool your worker serves - Ensure the worker’s PREFECT_API_URL points to the same API you’re cancelling against - Print the run’s infrastructure_pid (namespace:job_name) and verify that Job exists in that namespace:

kubectl -n <ns_from_pid> get job <job_from_pid> -o name

Docs for reference: - Kubernetes worker how‑to: https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/ - Settings reference: https://docs.prefect.io/v3/api-ref/settings-ref/

Totally get it. Default fallback polling is 120 seconds. Details: - The worker’s cancellation observer uses the events stream first (near-instant). Polling only kicks in if the events WebSocket isn’t available. - Default polling interval:

PREFECT_WORKER_CANCELLATION_POLL_SECONDS=120

- Setting it to 60 (or even 30) is fine if you need faster pickup when events aren’t available; just be mindful it increases API traffic a bit. What I recommend:

PREFECT_WORKER_ENABLE_CANCELLATION=true
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60
PREFECT_LOGGING_LEVEL=DEBUG  # temporarily, to confirm observer starts

Then inside the worker pod, verify:

prefect config view | grep -i worker.enable_cancellation
prefect config view | grep -i cancellation_poll_seconds

If your events path is healthy (WebSocket allowed through your ingress), cancels should be near real-time regardless of the polling value. If events aren’t available, the poll interval is your reaction time. Refs: - Kubernetes worker how-to: https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/ - Settings reference: https://docs.prefect.io/v3/api-ref/settings-ref/

You didn’t disable events — but cancellation isn’t “just events.” Two separate things are at play: - Events stream: near‑instant notifications from the Prefect API (via WebSocket) - Worker cancellation observer: the component that listens for those events (or polls if events aren’t reachable) and actually kills the Kubernetes Job/Pod If

PREFECT_WORKER_ENABLE_CANCELLATION

is false, the worker won’t run the cancellation observer at all. So even if events are available, nothing will act on them. What to do now - Ensure the worker is actually running the cancellation observer:

PREFECT_WORKER_ENABLE_CANCELLATION=true
# optional fallback polling if events aren’t reachable
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60
PREFECT_LOGGING_LEVEL=DEBUG   # temporarily, to verify behavior

- Restart the worker and watch startup logs. You should see it start a FlowRunCancellingObserver and either: - subscribe to the events stream (instant cancels), or - fall back to polling every N seconds. How to verify quickly 1) Inside the worker pod:

prefect config view | grep -i worker.enable_cancellation
prefect config view | grep -i cancellation_poll_seconds

2) Cancel a test run in the same work pool the worker serves. With events working, you’ll see cancel logs almost immediately; if not, you’ll see them within the poll interval. If cancels still aren’t picked up immediately - The worker may not be able to open the WebSocket to the Prefect API (ingress/load balancer/proxy blocking “Connection: Upgrade” / “Upgrade: websocket”). In that case, it’ll fall back to polling — which will still work, just on the poll cadence. - Double-check the run’s work_pool_name matches the worker’s pool and that the worker’s PREFECT_API_URL points at the same API you’re clicking Cancel in. Docs for reference: - Kubernetes worker how‑to: https://docs.prefect.io/v3/how-to-guides/workers/kubernetes/ - Settings reference: https://docs.prefect.io/v3/api-ref/settings-ref/

I hear you. Quick clarification that should unblock you: - The “Kubernetes observer” you enabled (the kopf-based integration) is just for mirroring K8s events into Prefect’s event stream. It does not kill pods or handle flow-run cancels. - The thing that actually stops your Jobs/Pods on Cancel is the worker’s FlowRunCancellingObserver. That is part of core Prefect, not kopf, and it only runs if you set

PREFECT_WORKER_ENABLE_CANCELLATION=true

. What I recommend right now 1) Enable worker-side cancellation and verify it’s running:

# in your worker container env
PREFECT_WORKER_ENABLE_CANCELLATION=true
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60   # optional, for faster fallback
PREFECT_LOGGING_LEVEL=DEBUG                   # temporarily

- Restart the worker and watch logs at startup; you should see it initialize a FlowRunCancellingObserver (it’ll say it’s subscribing to events or falling back to polling). - Cancel a test run in the same work pool and you should see logs like “received cancellation for flow run …” and “deleting Kubernetes job …”. 2) Sanity checks if you still don’t see cancel logs - Confirm the run’s pool matches the worker’s pool and the worker’s API URL matches the UI you’re cancelling in. - Print infra_pid and verify it points to a real Job:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<RUN_ID>")
    print(fr.work_pool_name, fr.infrastructure_pid)  # ns:jobname

kubectl -n <ns_from_pid> get job <job_from_pid> -o name

About kopf not starting (optional, separate from cancellation) - Not required for cancels. If you still want it running: - Ensure your image includes the integration: install

prefect-kubernetes

(or

prefect[kubernetes]

). Inside the worker:

python -c "import prefect_kubernetes, kopf; print('ok')"

should succeed. - RBAC must include watch on events, pods, pods/log, and jobs in the namespaces you configured:

- apiGroups: [""]
      resources: ["pods","pods/log","events"]
      verbs: ["get","list","watch"]
    - apiGroups: ["batch"]
      resources: ["jobs"]
      verbs: ["get","list","watch"]

- Make sure

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES

resolves to a real namespace value (not a literal

${NAMESPACE}

if your Helm chart didn’t substitute it). - Start with DEBUG logs to see any kopf startup errors. Key point: kopf not starting won’t block cancellation. The worker’s own cancellation observer is what deletes the Job/Pod. Turn that on and you should see cancels take effect. If you want, paste the worker’s DEBUG startup lines after enabling

PREFECT_WORKER_ENABLE_CANCELLATION

, and I’ll confirm it’s wired correctly.

Here’s the concise rundown of 0.7.5 → 0.7.6 → 0.7.7 for prefect-kubernetes, focusing on things that could affect what you’re seeing: Notable changes in 0.7.6 (big feature drop) - Annotations support: you can set annotations in the work pool vars; they’re propagated to the Pod template. - Standard app.kubernetes.io labels: worker adds managed-by/part-of/version/name labels to Jobs/Pods in addition to Prefect’s own labels. - Pod failure diagnostics: observer now diagnoses common failures (ImagePullBackOff, CrashLoopBackOff, OOMKilled, Evicted, Unschedulable) and emits actionable messages. - Crashed pod logs forwarding: observer will fetch and forward logs for pods that crashed before connecting to Prefect. - InfrastructurePending proposals: if a pod is Pending while the run is Running, observer proposes InfrastructurePending for clearer UI. - Resource fields in work pool: cpu/memory request/limit fields added to variables. - Tunable job-creation retries: tenacity settings made configurable (no more hardcoded retry timings). - include_files support in kubernetes decorator: bundle config files into the flow package. - Fixes to observer’s Kubernetes API handling: switched to correct camelCase fields (containerStatuses, restartCount), added initContainerStatuses. 0.7.7 (bugfixes) - Preserve label key casing: custom label keys are no longer lowercased. - Kopf observer selector fix: explicitly scopes kopf to batch/v1 Jobs to avoid ambiguous resource matches. What this means for you - None of these versions change the core cancellation mechanism: the worker still deletes the Job using the flow run’s infrastructure_pid (namespace:job_name). Labels are not used for the delete operation. - If you rely on the observer for better status/logs, 0.7.6/0.7.7 improve visibility and reduce false negatives (e.g., correct containerStatuses handling, log forwarding on early crashes). - If you had custom labels that were being lowercased in 0.7.6, 0.7.7 fixes that. - Kopf not starting: 0.7.7’s selector fix can help if kopf previously errored on ambiguous “jobs” matches. If you want, I can suggest exactly which version to pin for stability in your cluster and how to configure the new diagnostics/log-forwarding settings to get better signals during startup failures.

Thanks for spelling out the symptoms — that helps a lot. Here’s the core of what’s going on and how to fix it: What’s actually responsible for cancelling - Dashboard Cancel always updates the run to Cancelling on the server. - For runs that have NOT started (no start_time): the worker can intercept and delete the Kubernetes Job before the engine starts — that’s what

_cancel_run

is for. The guard you found (

if flow_run.start_time is not None: return

) is by design so the worker doesn’t race the engine once it’s running. - For runs that HAVE started (start_time is set): the flow process inside the pod (the runner) is responsible for reacting to Cancelling and exiting. The Kubernetes kopf observer is not used for cancelling; it’s for mirroring K8s events and adding diagnostics. Why your running pods don’t stop - Your workers not starting kopf is unrelated to cancelling; it shouldn’t block cancellations. - The fact that running pods aren’t stopping means the runner inside the pod is not receiving/acting on Cancelling. Common reasons: - The runner cannot reach the events WebSocket and never falls back to polling properly (ingress/LB blocks WS upgrade), or it’s not connected to the same API you’re clicking Cancel in. - The pod image/env does not have valid Prefect connection settings at runtime (PREFECT_API_URL and, for Cloud, PREFECT_API_KEY) — it may start but lose connectivity later. - The flow code is in a tight/blocking loop that doesn’t cooperatively yield; the engine sees Cancelling but cannot interrupt the user code promptly. What to do now (minimal, targeted steps) 1) Verify the runner is actually connected to the same API and is reacting to Cancelling: - Temporarily run the flow pod with debug logging:

PREFECT_LOGGING_LEVEL=DEBUG

- Cancel a run and look in the pod logs for messages indicating cancellation observation (you should see logs about receiving Cancelling and shutting down). If you see nothing, the runner is not observing cancels. 2) Confirm the pod can reach the events stream (or at least poll): - From inside a flow pod, quickly test API access:

python - <<'PY'
import asyncio
from prefect import get_client
async def main():
    async with get_client() as c:
        fr = await c.read_flow_run("REPLACE_WITH_THIS_RUN_ID")
        print(fr.id, fr.state_name)
asyncio.run(main())
PY

If this fails or hangs, the runner cannot reach your Prefect API reliably; cancels won’t be observed. 3) Make cancellation of not-yet-started runs reliable: - Ensure the worker is allowed to intercept and kill pre-start runs:

PREFECT_WORKER_ENABLE_CANCELLATION=true
# optional, faster fallback polling if WS is blocked
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60

This won’t kill already-running pods (by design) but will stop queued/pending ones consistently. 4) Reduce kopf load (not required for cancelling but good for stability): - Prune old Jobs with TTL (you have ttlSecondsAfterFinished; verify the TTL controller is enabled and actually cleaning up). - Scope the observer to only necessary namespaces and (ideally) only Prefect-managed jobs. With many thousands of completed Jobs, kopf can stall its initial list/watch. - 0.7.7’s fix scoping kopf to batch/v1 helps with noisy matches; you’re already on it. If you absolutely need force-kill semantics for running pods via the worker - Prefect’s worker intentionally avoids deleting Jobs for runs that have started; it relies on the runner in the pod to exit cleanly on Cancelling. - As a temporary workaround, you could run a tiny service that listens for

prefect.flow-run.Cancelling

events and deletes the Kubernetes Job by its

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

label — but that’s a custom add-on, not built-in. - The better fix is ensuring the runner inside the pod observes Cancelling (WS access or polling) and your code cooperatively yields so the engine can stop it. Sanity checks to share so I can pinpoint the gap

- In the UI/API: does the run flip to Cancelling immediately after you click Cancel? - A few lines from the pod logs (with DEBUG) right after you click Cancel. - Confirm your pod has PREFECT_API_URL (and PREFECT_API_KEY if Cloud) and that it points to the same API you’re using. - Whether your flows involve long, non-yielding work (tight loops, heavy C extensions) that could ignore cooperative cancellation. I know this is frustrating — but the path is: worker cancels pre-start runs; the pod’s runner cancels running ones. Let’s get the runner observing Cancelling and you’ll see immediate improvement.

No. Prefect 3 does not use Redis as a broker/queue for scheduling or cancellation. The control plane is the Prefect API + Postgres; workers pull runs from the API, and cancellation signals are delivered via the API’s events stream (WebSocket) with API polling as fallback. Adding Redis won’t change cancel behavior. To get cancels working reliably: - Make sure worker-side cancellation is enabled and visible in the worker pod:

PREFECT_WORKER_ENABLE_CANCELLATION=true
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60   # fallback if WS isn’t reachable
PREFECT_LOGGING_LEVEL=DEBUG                   # temporarily, to verify

- Ensure your flow pods can reach the same Prefect API you’re using (PREFECT_API_URL [+ PREFECT_API_KEY if Cloud]) so the runner inside the pod can observe Cancelling and exit. - If the kopf observer is getting stuck on startup (lots of old Jobs), it’s safe to disable it; it’s not required for cancellation:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

And/or ensure TTL cleanup for finished Jobs is actually working so kopf isn’t flooded. References: - Kubernetes worker how-to - Settings reference

Totally fair to want the full picture. Here’s how cancellation delivery works in Prefect 3, what multiple API replicas mean, and how your worker pod settings relate. How cancellation is delivered - Two layers are involved: 1) The Prefect API persists state changes in Postgres. When you click Cancel, the run’s state becomes Cancelling on the server (durable). 2) The runtime that launched the work reacts: - Runner in the flow pod (in‑process): subscribes to the API’s events stream (WebSocket) for near‑instant Cancelling signals and, if that’s not reachable, falls back to periodic API checks. When it observes Cancelling, it exits the flow process cleanly. - Worker (external launcher): can optionally intercept and kill infrastructure before the process starts (pending runs) if you set PREFECT_WORKER_ENABLE_CANCELLATION=true; for already‑running pods, the worker does not forcibly delete the Job by default—the pod’s runner should exit when it sees Cancelling. Multiple API replicas - Not a problem by itself. The runner maintains a long‑lived WebSocket to one replica via your load balancer. If that connection drops, the runner reconnects. Even if an event is missed, the server’s state is the source of truth; the runner’s fallback polling still sees the Cancelling state. - What can break cancels across replicas is networking/ingress that blocks or aggressively times‑out WebSockets. If the WS cannot be established, you’ll only get polling‑cadence cancels. If both WS and polling are blocked, cancels won’t be observed. - NGINX Ingress: ensure websocket upgrade and sane timeouts - proxy_set_header Upgrade $http_upgrade; - proxy_set_header Connection "upgrade"; - proxy_read_timeout and proxy_send_timeout >= 3600s (or at least > your poll interval) - AWS ALB/NLB: idle timeout high enough (>= 300s is common) for WS longevity. Do those worker env vars affect cancellation? - The ones you showed are for the optional Kubernetes “observer” integration (kopf) that mirrors K8s Job/Pod events into Prefect and adds diagnostics/logs. They do not handle flow-run cancellation. - These settings do not interfere with the runner’s ability to observe Cancelling in the flow pod. - If kopf is hanging on startup due to a massive initial watch (lots of old Jobs), you can safely disable it—it’s unrelated to cancel mechanics: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false What you do need set for worker-side pre-start cancels - To allow the worker to kill not‑yet‑started Jobs (pending runs): - PREFECT_WORKER_ENABLE_CANCELLATION=true - (optional) PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60 for faster fallback if WS is blocked What the flow pod needs for running-runs cancels - Valid Prefect connection in the container: - PREFECT_API_URL (and PREFECT_API_KEY if Cloud), pointing at the same API you use in the UI - Outbound network that allows: - HTTPS to the API - WebSocket upgrades to the API (preferred). If WS can’t be used, periodic API polling still works—just slower. About Redis in the docs - Prefect 3’s self‑hosted stack does not use Redis as a broker/queue for scheduling or cancellation. Control plane = Prefect API + Postgres. Workers pull runs from the API; cancellations travel via events WS with API polling fallback. - If you saw Redis mentioned, it’s either: - an older doc/diagram from Prefect 1.x, or - an optional component in someone’s example stack (e.g., for other purposes). It is not required or used by Prefect 3 for cancels. Quick verification checklist - In a running flow pod: - Confirm it can hit the API: - Try a quick client call (read_flow_run) or check logs for “connected” vs “falling back to polling”. - Cancel the run from the UI: - With WS working: expect near‑instant shutdown. - Without WS: expect shutdown on the poll cadence. - In the worker pod:

- If you want pre‑start cancels, set PREFECT_WORKER_ENABLE_CANCELLATION=true and watch DEBUG logs for the cancellation observer at startup. - In ingress/LB: - Ensure WebSocket upgrade and adequate idle/timeouts so long‑lived WS connections aren’t dropped prematurely. If you want, share: - How your ingress/LB is set up (NGINX/ALB and timeouts) - One flow pod’s env for PREFECT_API_URL(/KEY) - A short excerpt of that pod’s logs after you click Cancel (DEBUG helps) I’ll map that to the exact change (WS headers/timeouts, enabling worker cancellation for pending runs, or disabling kopf if it’s just adding noise) so running pods actually stop when you cancel.

I get why this is confusing, so here’s the exact picture and what to change. First: Redis is not part of Prefect 3’s orchestration path. There is no Redis broker for scheduling or cancellations. The cancel path is: - You click Cancel in the UI → the Prefect API (backed by Postgres) records the run as Cancelling. - The runtime that launched work reacts: - In-process runner (inside your flow pod): listens to the Prefect API’s events stream over WebSocket and, if that’s not reachable, falls back to polling the API. When it sees Cancelling, it exits the flow process. - Worker (external launcher): can optionally intercept and kill infrastructure for runs that have not started yet (pending) if you set PREFECT_WORKER_ENABLE_CANCELLATION=true. By design, it does not forcibly delete Jobs for runs that are already running. Multiple API replicas do not “lose” cancels: - Your flow pods connect to the internal ClusterIP service. That’s fine—no ALB involved, and there’s no idle timeout on a ClusterIP. - The runner keeps a WebSocket to one API replica. If WS drops, it reconnects. Even if a single event were missed, the server’s Cancelling state is durable in Postgres and the runner’s fallback polling sees it. Those Kubernetes observer (kopf) env vars you set are unrelated to cancelling: - PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_* controls an optional observer that mirrors K8s Job/Pod events into Prefect and adds diagnostics. It does not handle flow-run cancellations. - If kopf is stuck on startup due to a huge backlog of old Jobs, you can disable it; cancelling will still work once the runner/worker observers are correctly configured. What to change now - Ensure pre-start (pending) cancels work: enable the worker’s cancellation observer.

PREFECT_WORKER_ENABLE_CANCELLATION=true
PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60   # optional, faster fallback if WS isn’t available
PREFECT_LOGGING_LEVEL=DEBUG                   # temporarily to confirm behavior

- Ensure running-runs cancels work: the flow pod’s runner must observe Cancelling. - The pod needs valid Prefect connection env at runtime pointing to the same API: - PREFECT_API_URL (and PREFECT_API_KEY for Cloud) - Temporarily crank logs to confirm the runner is observing cancels and will exit:

PREFECT_LOGGING_LEVEL=DEBUG
    PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE=true

With this set, if the runner cannot start its cancellation observer (WS + polling), it will crash loudly instead of silently continuing—useful for diagnosis. Sanity checks to prove the path - After clicking Cancel, verify the server accepted it:

from prefect import get_client
async with get_client() as c:
    fr = await c.read_flow_run("<RUN_ID>")
    print(fr.state_name, fr.state_type)  # Expect: Cancelling → Cancelled

- In the flow pod (with DEBUG), cancel a test run and watch for logs indicating it received Cancelling and is shutting down. If you see nothing, the runner is not observing cancels (WS blocked or polling not happening). - In the worker pod (with DEBUG) and worker cancellation enabled, cancel a pending run; you should see it log “received cancellation…” and “deleting Kubernetes job …”. About kopf and backlogs - If kopf is hanging on startup because of thousands of old Jobs, that won’t block cancellations (it’s a separate integration). Either: - disable it:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

- or ensure TTLSecondsAfterFinished is actually cleaning completed Jobs so kopf’s initial watch isn’t overwhelmed. If you truly need force-kill of already running pods from outside the pod: - Prefect’s worker intentionally doesn’t do that by default. If it’s a hard requirement, the pragmatic workaround is a small controller that listens for prefect.flow-run.Cancelling events and deletes the Job by label. But that’s a custom add-on; the built-in path expects the runner in the pod to observe Cancelling and exit.

Bottom line - Don’t add Redis—it won’t change cancellations in Prefect 3. - Enable worker cancellation for pending runs. - Make sure the flow pod’s runner can observe Cancelling (WS preferred, polling fallback) and that PREFECT_API_URL(/KEY) are correct in the pod. - Disable kopf if it’s just adding noise—cancels don’t depend on it. If you share: - A flow pod’s env for PREFECT_API_URL(/KEY) - A short DEBUG excerpt from that pod right after you click Cancel - The worker pod’s

prefect config view | grep -i cancellation

output …I’ll point to the exact gap (runner observer not starting, API URL mismatch, or worker cancellation still disabled) and give you a precise fix.

I hear you. But to be precise: Redis is not a requirement for Prefect 3’s self‑hosted control plane, scheduling, or cancellation. Prefect 3 core uses the API + Postgres. Workers pull from the API; running pods observe Cancelling via the API’s events WebSocket with API polling as fallback. There is no Redis broker in that path. It’s possible the page you’re referencing mentions Redis in a different context (e.g., optional caching layers or components in a reference stack), but it is not required for cancellations or work‑pool dispatch. If you can paste the exact excerpt from How to scale self-hosted Prefect, I’ll reconcile it (and get docs fixed if it’s misleading). Recap of the cancellation mechanics (Prefect 3): - Server: clicking Cancel persists Cancelling in Postgres. - Running pod: the runner inside the pod: - Subscribes to the API’s events WS; if WS isn’t reachable, it polls the API. - On seeing Cancelling, it exits the flow. - Worker (Kubernetes): optionally intercepts and kills not‑yet‑started runs if you set:

PREFECT_WORKER_ENABLE_CANCELLATION=true
  PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60  # optional, fallback cadence

By design, the worker does not force‑delete Jobs for runs that have already started; the pod’s runner should exit on Cancelling. Re your current setup: - The kopf observer is not used for cancellation. If it’s getting stuck on initial list/watch, you can safely disable it:

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

- Since your flow pods connect via ClusterIP to the API (no ALB in path), WS should be fine. If cancels still aren’t observed: - Verify pod env points at the same API: -

PREFECT_API_URL

(and

PREFECT_API_KEY

if Cloud) - Temporarily enable debug/crash‑on‑failure in the pod to confirm the runner’s cancellation observer is alive:

PREFECT_LOGGING_LEVEL=DEBUG
    PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE=true

- After clicking Cancel, you should see logs in the pod indicating it received Cancelling and is shutting down. If you see nothing, the runner isn’t observing (WS blocked or polling not happening). If you share the doc excerpt that mentions Redis, I’ll address it line‑by‑line. And if you want, I can give you a quick test recipe (tiny flow + test worker env) to prove cancellations work without Redis in your cluster.

You’re right to quote that doc. I need to correct and clarify: - In single‑server mode, Prefect 3 does not require Redis. - In multi‑server (multiple API replicas) self‑hosted mode, Redis is required for event fan‑out between replicas. That includes delivery of Cancelling events to long‑lived WebSocket subscribers. Without Redis, each API pod only emits events to its own in‑memory bus, so a flow pod connected to replica A may never see a cancel that was processed/emitted by replica B. That matches what you’re seeing: runs flip to Cancelling in the DB, but the running pod never receives the Cancelling event over its WS because there’s no cross‑replica event bus. Kopf is unrelated to cancellations here. What I recommend - Implement Redis for the API’s event messaging per the doc you linked: How to scale self-hosted Prefect - Prefect - Configure the Prefect API to use your Redis (ElastiCache) as the events backend per that guide - Restart API pods to pick up the config - After this, Cancelling events will be replicated across all API replicas; flow pods connected to any replica will receive them immediately over WS - As a quick validation path: - Temporarily scale the API to 1 replica and test cancel → the running pod should observe Cancelling quickly (proves WS + runner path works without cross‑replica fan‑out) - Keep these also in place: - For pre‑start (pending) cancels: set

PREFECT_WORKER_ENABLE_CANCELLATION=true

(worker deletes the Job before the engine starts) - If kopf is getting stuck on an initial flood of old Jobs, you can safely disable it; it’s not part of the cancel path: -

PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED=false

- Ensure TTL cleanup for finished Jobs is actually happening to avoid future floods Sanity checklist after enabling Redis - Start a run, confirm the pod connects to your ClusterIP service (as you described) - Click Cancel in the UI - In the pod logs, you should see the runner receive Cancelling almost immediately and begin shutdown - If you want faster fallback when WS can’t be reached (e.g., during maintenance), you can still set: -

PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60

(worker fallback for pending runs) - Optionally, temporarily add

PREFECT_LOGGING_LEVEL=DEBUG

to the worker or pod to observe the behavior If you want, I can review the Redis config stanza you plan to apply to the API Deployment and confirm it lines up with the doc so events are actually fanned out across your replicas.

You’re right, and I’m sorry. At your scale with multiple API replicas, Redis is required for cross-replica event fan-out. Without it, Cancelling events are only emitted in-memory on the replica that processed the request; pods connected to a different replica won’t see them. That explains exactly why your running pods didn’t stop. Thanks for calling that out. What to lock in - Keep Redis configured as the API’s event backend so Cancelling (and other) events reach all subscribers across replicas. - The kopf observer is optional and unrelated to cancels; feel free to keep it off if it’s noisy. - Keep worker-side cancellation enabled to catch not-yet-started runs: - PREFECT_WORKER_ENABLE_CANCELLATION=true - (optional) PREFECT_WORKER_CANCELLATION_POLL_SECONDS=60 Quick verification - With Redis + >1 API replica, start a run, then Cancel: - Pod logs should show near-instant “received cancellation” and graceful shutdown. - As a control, scale API to 1 replica and confirm cancels also work (WS + DB only). - Monitor Redis health (connections, mem, latency) to avoid event backlogs under bursts. Docs - The requirements you quoted are correct for multi-server: How to scale self-hosted Prefect If you’d like, share your API Deployment/values for the Redis wiring and I’ll sanity-check them and suggest alerting thresholds so this stays solid under peak load. And again—apologies for the earlier misdirection.