Great question! There are two common patterns for cleaning up Kubernetes Jobs with Prefect. Assuming you’re on Prefect 3.x, here are the recommended options: Option 1 (recommended): Let Kubernetes auto-delete finished Jobs (TTL) - Prefect’s Kubernetes worker templates already include

spec.ttlSecondsAfterFinished

. You can set this via your work pool’s base job template so K8s automatically deletes Jobs after they finish. - This is zero-code and the most reliable approach. How to set it: 1) Fetch the default template

prefect work-pool get-default-base-job-template --type kubernetes > k8s.json

2) Edit k8s.json and set a TTL (seconds). For immediate deletion use 0; for 5 minutes use 300. You’ll see this in the job manifest:

"spec": {
  ...
  "ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
}

Then set the variable under the template’s variables section, e.g.:

"variables": {
  ...
  "finished_job_ttl": 300
}

3) Apply the template to your pool

prefect work-pool update <your-pool-name> --base-job-template k8s.json

Docs: - Kubernetes worker/pool how-to - prefect-kubernetes integration Option 2: Run a Prefect “janitor” flow that deletes completed Jobs If you prefer a Prefect flow to clean up Jobs (e.g., on a schedule or with extra logic), you can use the prefect-kubernetes collection: Example janitor flow:

from datetime import datetime, timezone, timedelta

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from <http://prefect_kubernetes.jobs|prefect_kubernetes.jobs> import list_namespaced_job, delete_namespaced_job
from kubernetes.client import V1DeleteOptions

@flow
def cleanup_completed_jobs(
    namespace: str = "default",
    older_than_minutes: int = 30,
    name_prefix: str = "prefect",  # adjust to your naming
    cascade: bool = True,
):
    creds = KubernetesCredentials.load("my-k8s-creds")  # or configure in-cluster
    job_list = list_namespaced_job(kubernetes_credentials=creds, namespace=namespace)
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)

    delete_opts = V1DeleteOptions(propagation_policy="Foreground") if cascade else None

    for job in job_list.items:
        name = job.metadata.name or ""
        created = job.metadata.creation_timestamp
        succeeded = (job.status.succeeded or 0) >= 1

        if not succeeded:
            continue
        if created and created >= cutoff:
            continue
        if not name.startswith(name_prefix):
            continue

        delete_namespaced_job(
            kubernetes_credentials=creds,
            job_name=name,
            namespace=namespace,
            delete_options=delete_opts,
        )

- Schedule this flow to run periodically. - Narrow your deletion criteria carefully (namespace, name prefix, or custom labels) to avoid deleting non-Prefect jobs. Option 3: “Cleanup task” at the end of your flow - Prefect flows support `on_completion`/`on_failure` hooks, but deleting the Kubernetes Job from within the same Job is brittle and not generally recommended. The TTL approach is much safer and simpler. - Use in-flow cleanup for child resources your flow creates (temporary pods, services, etc.), not for the Job running the flow itself. Quick check: Are you trying to clean up - the Kubernetes Job that runs each Prefect flow (worker-created), or - Jobs you launch from inside your flows? If you share which one you need (plus your namespace and desired retention time), I can tailor the exact JSON or code snippet for your setup.

Got it — for Prefect 2.20.16 you’ve got a few good options. Short version: use Kubernetes’ TTL to auto-delete finished Jobs; if you can’t, run a small “janitor” flow with prefect-kubernetes to delete them. Recommended: Set Finished Job TTL on your Kubernetes work pool (workers) - This sets Kubernetes’ native

spec.ttlSecondsAfterFinished

so K8s automatically deletes finished Jobs after N seconds. - How to set it: - In the UI: Work Pools → your Kubernetes pool → set “Finished Job TTL (seconds)” → Save. - Or via CLI using the base job template: 1) Export the template

prefect work-pool get-default-base-job-template --type kubernetes > k8s.json

2) Edit k8s.json: set the variable

finished_job_ttl

(e.g., 300 for 5 minutes). The manifest already includes:

"spec": {
      ...
      "ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
    }

3) Update your pool

prefect work-pool update <your-pool-name> --base-job-template k8s.json

- Docs: - Work pools (2.x) - Kubernetes deployments (2.x) If you’re using agents with KubernetesJob infrastructure blocks (older 2.x pattern) - You can set TTL on the block itself:

from prefect.infrastructure.kubernetes import KubernetesJob

k8s = KubernetesJob(
    image="your-image:tag",
    finished_job_ttl=300,  # seconds
    # ... other fields ...
)
k8s.save("prod-k8s", overwrite=True)

- This translates to

ttlSecondsAfterFinished

in the Job manifest. - Note: Workers/work pools are preferred in 2.20.16; consider migrating if you’re still on agents. Cleanup via a Prefect “janitor” flow (if you need more control) - Use the prefect-kubernetes collection to list and delete Jobs, optionally filtered by labels, age, or status.

from datetime import datetime, timezone, timedelta

from prefect import flow
from prefect_kubernetes.credentials import KubernetesCredentials
from <http://prefect_kubernetes.jobs|prefect_kubernetes.jobs> import list_namespaced_job, delete_namespaced_job
from kubernetes.client import V1DeleteOptions

@flow
def cleanup_completed_jobs(
    namespace: str = "default",
    older_than_minutes: int = 30,
    label_selector: str = "app=prefect",  # adjust to your labels
    cascade: bool = True,
):
    creds = KubernetesCredentials.load("my-k8s-creds")  # or rely on in-cluster config
    job_list = list_namespaced_job(
        kubernetes_credentials=creds,
        namespace=namespace,
        label_selector=label_selector
    )

    cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)
    delete_opts = V1DeleteOptions(propagation_policy="Foreground") if cascade else None

    for job in job_list.items:
        created = job.metadata.creation_timestamp
        succeeded = (job.status.succeeded or 0) >= 1
        if not succeeded:
            continue
        if created and created >= cutoff:
            continue

        delete_namespaced_job(
            kubernetes_credentials=creds,
            job_name=job.metadata.name,
            namespace=namespace,
            delete_options=delete_opts,
        )

- Schedule this janitor flow to run periodically. - Docs: prefect-kubernetes (2.x) About “cleanup tasks” at the end of your flow - If you mean deleting the very Job that is running your flow: use Finished Job TTL instead of in-flow deletion (more reliable). - For resources your flow creates (child pods/jobs/services), you can use state change hooks to ensure cleanup even on failure: - State change hooks (2.x) Quick clarifier to tailor this further: - Are you using workers/work pools or agents/infrastructure blocks?

- Do you want to clean up the worker-created flow Jobs, or Jobs you launch from inside your flows? - Desired retention window (e.g., delete immediately, after 5 minutes, after 1 day)?

Great follow-up. In Prefect 2.20.16, cleaning up the Kubernetes Job does not automatically clean up volumes. You’ve got two solid approaches depending on what you’re using: Option A (preferred): Use ephemeral volumes so there’s nothing to clean up - emptyDir: Tied to the Pod lifecycle; removed when the Pod is deleted. - Generic ephemeral volumes (ephemeral volumeClaimTemplate): Kubernetes provisions a per-Pod PVC that is deleted when the Pod is deleted. How to add an ephemeral volume in your Kubernetes work pool’s base job template 1) Export your pool’s template (if you haven’t already)

prefect work-pool get-default-base-job-template --type kubernetes > k8s.json

2) In

k8s.json

, add a

volumes

entry and a matching

volumeMounts

to the container under `job_manifest.spec.template.spec`: emptyDir example (no PV/PVC created):

"spec": {
  "template": {
    "spec": {
      "volumes": [
        {
          "name": "workspace",
          "emptyDir": {}
        }
      ],
      "containers": [
        {
          "name": "prefect-job",
          "args": "{{ command }}",
          "env": "{{ env }}",
          "image": "{{ image }}",
          "imagePullPolicy": "{{ image_pull_policy }}",
          "volumeMounts": [
            {
              "name": "workspace",
              "mountPath": "/data"
            }
          ]
        }
      ],
      "restartPolicy": "Never",
      "serviceAccountName": "{{ service_account_name }}"
    }
  }
}

Generic ephemeral volume (PVC per Pod, deleted with Pod):

"spec": {
  "template": {
    "spec": {
      "volumes": [
        {
          "name": "workspace",
          "ephemeral": {
            "volumeClaimTemplate": {
              "spec": {
                "accessModes": ["ReadWriteOnce"],
                "resources": { "requests": { "storage": "10Gi" } },
                "storageClassName": "your-storage-class"
              }
            }
          }
        }
      ],
      "containers": [
        {
          "name": "prefect-job",
          "args": "{{ command }}",
          "env": "{{ env }}",
          "image": "{{ image }}",
          "imagePullPolicy": "{{ image_pull_policy }}",
          "volumeMounts": [
            { "name": "workspace", "mountPath": "/data" }
          ]
        }
      ],
      "restartPolicy": "Never",
      "serviceAccountName": "{{ service_account_name }}"
    }
  }
}

3) Update the pool

prefect work-pool update <your-pool-name> --base-job-template k8s.json

Kubernetes docs: - emptyDir - Generic ephemeral volumes Option B: If you’re using standalone PVCs, delete them explicitly - Deleting the Job will not delete PVCs. You can: - Ensure the StorageClass has reclaimPolicy=Delete so deleting the PVC also deletes the underlying PV (e.g., EBS). If reclaimPolicy=Retain, the PV will remain. - Run a small “janitor” Prefect flow on a schedule that deletes PVCs that match your labels/age criteria. Example janitor flow (uses Kubernetes Python client; run this as a Prefect flow): ``` from datetime import datetime, timezone, timedelta from prefect import flow from kubernetes import client, config @flow def delete_old_pvcs( namespace: str = "default", label_selector: str = "app=prefect", # add labels to your PVCs so they’re easy to select older_than_minutes: int = 30, ): # Works in-cluster; for local testing use config.load_kube_config() try: config.load_incluster_config() except Exception: config.load_kube_config() v1 = client.CoreV1Api() pvcs = v1.list_namespaced_persistent_volume_claim( namespace=namespace, label_selector=label_selector ) cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes) for pvc in pvcs.items: created = pvc.metadata.creation_timestamp

if created and created < cutoff: v1.delete_namespaced_persistent_volume_claim( name=pvc.metadata.name, namespace=namespace ) ``` Notes: - Label your PVCs (e.g.,

app=prefect

or a flow-specific label) when you create them so your janitor can target them safely. - If your StorageClass’s reclaimPolicy=Delete, deleting the PVC also deletes the PV/resource in your cloud provider. If reclaimPolicy=Retain, you’ll need to clean up PVs separately. - Finalizers or in-use claims can block deletion; ensure Jobs/Pods are finished first. Which are you using today? - A PVC you create per flow run? - A static PVC shared by many runs? - Or do you want ephemeral scratch space per run? Share a bit about your current volume setup (PVC vs emptyDir vs ephemeral, storage class, desired retention), and I can give you a tailored base job template or janitor flow that fits your cluster and storage class.

Totally doable on 2.20.16 without ephemeral volumes. The pattern is: run a scheduled “janitor” flow that mounts the same PVC and deletes files you no longer need. High-level - Create a dedicated Kubernetes work pool whose base job template mounts your PVC at a known path (e.g., /data). - Deploy a Prefect flow that deletes files older than N days from that path. - Schedule it to run once a day (end-of-day). - Keep the normal TTL cleanup for Jobs separate so the janitor Job also cleans itself up. 1) Mount your PVC in the pool’s base job template - Export and edit the pool template:

prefect work-pool get-default-base-job-template --type kubernetes > cleanup-pool.json

- Add your PVC and volumeMount to the job manifest (same namespace as the PVC):

"spec": {
  "template": {
    "spec": {
      "volumes": [
        {
          "name": "shared-data",
          "persistentVolumeClaim": { "claimName": "YOUR_PVC_NAME" }
        }
      ],
      "containers": [
        {
          "name": "prefect-job",
          "args": "{{ command }}",
          "env": "{{ env }}",
          "image": "{{ image }}",
          "imagePullPolicy": "{{ image_pull_policy }}",
          "volumeMounts": [
            { "name": "shared-data", "mountPath": "/data" }
          ]
        }
      ],
      "restartPolicy": "Never",
      "serviceAccountName": "{{ service_account_name }}"
    }
  }
}

- Optional (recommended): Set a Finished Job TTL so the cleanup Job gets removed automatically:

"spec": {
  ...
  "ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
}

and set

"finished_job_ttl": 300

under variables. - Save to a dedicated pool so you don’t mount the PVC on every Job:

prefect work-pool create "cleanup-pool" --type kubernetes --base-job-template cleanup-pool.json

Docs: Kubernetes (2.x) • Work pools (2.x) 2) Prefect cleanup flow that removes old files - This runs inside the Job, sees the PVC at /data, and deletes files older than N days.

import os, time
from prefect import flow, get_run_logger

@flow
def cleanup_volume(
    root_path: str = "/data",
    keep_days: int = 7,
    remove_empty_dirs: bool = True,
    dry_run: bool = False,
):
    logger = get_run_logger()
    cutoff = time.time() - keep_days * 86400
    deleted_files = 0
    deleted_dirs = 0
    scanned = 0

    for dirpath, dirnames, filenames in os.walk(root_path, topdown=False):
        for fname in filenames:
            fpath = os.path.join(dirpath, fname)
            scanned += 1
            try:
                mtime = os.path.getmtime(fpath)
            except FileNotFoundError:
                continue
            if mtime < cutoff:
                if dry_run:
                    <http://logger.info|logger.info>(f"[DRY RUN] Would delete: {fpath}")
                else:
                    try:
                        os.remove(fpath)
                        deleted_files += 1
                    except Exception as e:
                        logger.warning(f"Failed to delete {fpath}: {e}")

        if remove_empty_dirs:
            try:
                if not os.listdir(dirpath):
                    if dry_run:
                        <http://logger.info|logger.info>(f"[DRY RUN] Would remove empty dir: {dirpath}")
                    else:
                        os.rmdir(dirpath)
                        deleted_dirs += 1
            except Exception:
                pass

    <http://logger.info|logger.info>(
        f"Scanned {scanned} files; deleted {deleted_files} files; "
        f"removed {deleted_dirs} dirs (keep_days={keep_days})"
    )

3) Deploy and schedule it daily - In 2.20.16 you can: - Use the UI: create a Deployment for

cleanup_volume

, select the “cleanup-pool”, and add a daily schedule (e.g., 23:55 local). - Or use Python deployment (2.x-style) if you prefer code: - Note: in 3.x this method changes; this is correct for 2.x. ``` from prefect import flow

from prefect.deployments import Deployment from prefect.server.schemas.schedules import CronSchedule # assuming cleanup_volume is defined above if name == "__main__": Deployment.build_from_flow( flow=cleanup_volume, name="daily-volume-cleanup", work_pool_name="cleanup-pool", schedule=CronSchedule(cron="55 23 * * *", timezone="UTC"), parameters={"root_path": "/data", "keep_days": 7, "remove_empty_dirs": True, "dry_run": False}, ).apply() ``` Practical tips - Namespace/PVC: The Job’s namespace must match the PVC’s namespace. Use the same namespace in your work pool/template. - Permissions: If files are owned by a specific UID/GID, set a compatible security context (e.g.,

fsGroup

on the Pod or `runAsUser`/`runAsGroup` on the container) so deletes don’t fail. - Safety: Start with

dry_run=True

to confirm what will be removed, then flip to False. - Concurrency: Schedule cleanup when your pipeline is idle to avoid deleting actively-written files. If needed, use “last modified” plus a cushion (e.g., keep_days >= 1). - Alternatives: You can also run a Kubernetes CronJob outside Prefect that mounts the same PVC and runs

find /data -type f -mtime +7 -delete

if you prefer a pure K8s solution. If you share: - PVC name and namespace - Desired mount path - Your retention rule (e.g., delete files older than 7 days) I can produce a ready-to-use base job template and deployment config for your cluster.