Thang Le
03/18/2026, 1:59 PMMarvin
03/18/2026, 1:59 PMThang Le
03/18/2026, 2:08 PMMarvin
03/18/2026, 2:08 PMMarvin
03/18/2026, 2:16 PM{
"variables": {
"service_account_name": {
"title": "Service Account Name",
"type": "string",
"default": "prefect-runner"
},
"extra_annotations": {
"title": "Extra annotations",
"type": "object",
"default": {
"owner": "platform-team"
}
},
"extra_labels": {
"title": "Extra labels",
"type": "object",
"default": {
"team": "data-eng",
"env": "prod"
}
},
"pod_security_context": {
"title": "Pod security context",
"type": "object",
"default": {
"runAsNonRoot": true,
"runAsUser": 1000
}
}
},
"job_configuration": {
"job_manifest": {
"apiVersion": "batch/v1",
"kind": "Job",
"metadata": {
"annotations": "{{ extra_annotations }}",
"labels": "{{ extra_labels }}"
},
"spec": {
"template": {
"metadata": {
"annotations": "{{ extra_annotations }}",
"labels": "{{ extra_labels }}"
},
"spec": {
"serviceAccountName": "{{ service_account_name }}",
"securityContext": "{{ pod_security_context }}"
}
}
}
}
}
}
- Update an existing pool with a template:
prefect work-pool update "k8s-pool" --base-job-template ./base-job-template.json
- Get the default template as a starting point:
prefect work-pool get-default-base-job-template --type kubernetes
B) One work pool + multiple work queues + per-queue workers with env-based defaults
- Keep a single Kubernetes work pool (so no pool sprawl).
- Create multiple work queues with tag filters (e.g., team-a, team-b). Users just tag their deployments; they don’t touch job variables.
- Run one worker per queue, each with environment variables defining your defaults. In your base template, fall back to env vars:
"serviceAccountName": "{{ service_account_name | default($DEFAULT_SERVICE_ACCOUNT, true) | default('prefect-runner') }}"
- Then set the worker pod env, e.g.:Marvin
03/18/2026, 2:16 PM<http://app.kubernetes.io/managed-by|app.kubernetes.io/managed-by>: prefect, so you can easily target Prefect-run pods. Example Kyverno policy to inject EKS Security Groups for Pods:
apiVersion: <http://kyverno.io/v1|kyverno.io/v1>
kind: ClusterPolicy
metadata:
name: inject-eks-sgp
spec:
rules:
- name: add-eks-sgp
match:
resources:
kinds: ["Pod"]
labels:
<http://app.kubernetes.io/managed-by|app.kubernetes.io/managed-by>: "prefect"
mutate:
patchStrategicMerge:
metadata:
annotations:
<http://vpc.amazonaws.com/security-groups|vpc.amazonaws.com/security-groups>: "sg-0123456789abcdef,sg-abcdef0123456789"
This lets you keep Prefect generic while platform policy injects the right SA/SGP/labels automatically.
Other notes and gotchas
- Jinja context in templates has ctx.flow and ctx.flow_run. You can use flow_run.tags for simple conditionals if you adopt a tagging convention.
- Templates can reference environment variables with {{ $ENV_VAR }}.
- Work queues cannot define their own defaults/templates; all schema/defaults come from the work pool’s base job template and are overridden at deployment or run-time only.
- Prefect automatically adds some labels; your labels/annotations are merged, not replaced.
Questions to tailor a solution
- Are you on EKS and specifically using IRSA and Security Groups for Pods? If yes, are admission controllers (Kyverno/Gatekeeper) an option in your cluster?
- Would a single work pool + multiple queues + one worker per queue with env-based defaults meet your needs (i.e., users only add a tag; no job variable edits)?
- If you prefer strictly one queue and one worker, do you want to drive defaults via a tagging convention and Jinja conditionals in the template?
If you share your constraints, I can sketch the exact base job template and queue setup (or a Kyverno policy) for your environment.Marvin
03/18/2026, 2:17 PMjob_configuration with {{ variable_name }}.
- Precedence (lowest → highest): pool template defaults < deployment job_variables < run-time --job-variable.
- The name field supports {{ ctx.flow.* }} and {{ ctx.flow_run.* }} when the job is prepared for a run (handy for naming).
Minimal patterns and examples
- Generic structure
{
"variables": {
"image": {
"type": "string",
"default": "prefecthq/prefect:3-latest"
},
"env": {
"type": "object",
"default": {}
}
},
"job_configuration": {
"image": "{{ image }}",
"env": "{{ env }}",
"name": "{{ ctx.flow.name }}-{{ ctx.flow_run.id }}"
}
}
- Docker pool: make image and env configurable
{
"variables": {
"image": {
"type": "string",
"default": "my-image:latest"
},
"env": {
"type": "object",
"default": {
"LOG_LEVEL": "INFO"
}
}
},
"job_configuration": {
"image": "{{ image }}",
"env": "{{ env }}"
}
}
- Kubernetes pool: parameterize resources (snippet of job_manifest)
{
"variables": {
"cpu_request": {
"type": "string",
"default": "200m"
},
"memory_request": {
"type": "string",
"default": "512Mi"
}
},
"job_configuration": {
"job_manifest": {
"spec": {
"template": {
"spec": {
"containers": [
{
"name": "prefect-job",
"resources": {
"requests": {
"cpu": "{{ cpu_request }}",
"memory": "{{ memory_request }}"
}
}
}
]
}
}
}
}
}
}
- Process pool: working directory and env
{
"variables": {
"working_dir": {
"type": "string",
"default": null
},
"env": {
"type": "object",
"default": {}
}
},
"job_configuration": {
"working_dir": "{{ working_dir }}",
"env": "{{ env }}"
}
}
Override at deploy/run time
- In prefect.yaml (deployment-level)
deployments:
- name: my-deployment
entrypoint: flows/my_flow.py:my_flow
work_pool:
name: my-pool
job_variables:
image: "my-image:1.2.3"
env:
LOG_LEVEL: "DEBUG"
- At run time (highest precedence)
prefect deployment run my-flow/my-deployment \
--job-variable image='"my-image:hotfix"' \
--job-variable env.LOG_LEVEL='"TRACE"'
Useful CLI helpers (verified)
- Dump a default template you can edit:
prefect work-pool get-default-base-job-template --type docker > docker-base.json
prefect work-pool get-default-base-job-template --type kubernetes > k8s-base.json
prefect work-pool get-default-base-job-template --type process > process-base.json
- Create/update a pool with your template:
prefect work-pool create "my-pool" --type kubernetes --base-job-template ./k8s-base.json
prefect work-pool update "my-pool" --base-job-template ./k8s-base.json
Docs you can reference
- Work pools (concepts)
- Manage work pools (base job templates)
- Customize/override job variables
- Run flows in Docker
- Run flows on Kubernetes
- Run flows in local processes
If you tell me what you want to parameterize (e.g., image tag, service account, resources, working dir, env), I can sketch the exact base job template for your pool.Thang Le
03/18/2026, 2:25 PMan internal cli and the Prefect SDK for deployments in our playground environment. The problem is that the users using our cli and the prefect.client.orchestration package. We want to make sure both of these systems are the same as much as possibleMarvin
03/18/2026, 2:25 PMMarvin
03/18/2026, 2:27 PMserviceAccountName, annotations, labels, and any pod securityContext in the pool’s base job template.
- Define them as variables with sensible defaults and reference them in job_configuration with {{ variable_name }} so they apply to every job by default.
- This guarantees consistency regardless of how the deployment was created (prefect.yaml, SDK, or UI) as long as they target the same work pool.
2) Let teams override via deployment or run-time job variables (when needed)
- Deployment path (prefect.yaml) sets work_pool.job_variables to override pool defaults.
- SDK path should do the same with job_variables on .deploy(...) — identical keys → identical behavior.
3) Optionally, use Prefect Variables or env vars for environment-specific defaults
- In your base job template, you can reference Prefect Variables: {{ prefect.variables.default_service_account }}.
- Or environment variables: {{ $DEFAULT_SERVICE_ACCOUNT }}.
- This allows you to change defaults per environment (playground vs prod) without touching deployments.
4) Consider admission controllers for global org policy
- For things like IRSA/SGP/labels that must always be present, Kyverno/Gatekeeper/mutating webhooks ensure consistent injection no matter what Prefect does.
- Match on <http://app.kubernetes.io/managed-by=prefect|app.kubernetes.io/managed-by=prefect> to target Prefect-run pods.
Concrete examples
A) Base job template snippet (Kubernetes) — make SA, SGP annotation, and labels defaultable
{
"variables": {
"service_account_name": {
"title": "Service Account Name",
"type": "string",
"default": "{{ prefect.variables.default_service_account }}"
},
"annotations": {
"title": "Annotations",
"type": "object",
"default": {
"<http://vpc.amazonaws.com/security-groups|vpc.amazonaws.com/security-groups>": "{{ prefect.variables.default_sg_list }}"
},
"additionalProperties": {
"type": "string"
}
},
"labels": {
"title": "Labels",
"type": "object",
"default": {
"owner": "platform",
"env": "{{ prefect.variables.default_env }}"
},
"additionalProperties": {
"type": "string"
}
}
},
"job_configuration": {
"job_manifest": {
"apiVersion": "batch/v1",
"kind": "Job",
"metadata": {
"annotations": "{{ annotations }}",
"labels": "{{ labels }}"
},
"spec": {
"template": {
"metadata": {
"annotations": "{{ annotations }}",
"labels": "{{ labels }}"
},
"spec": {
"serviceAccountName": "{{ service_account_name }}",
"containers": [
{
"name": "prefect-job",
"env": "{{ env }}",
"args": "{{ command }}"
}
]
}
}
}
}
}
}
- Update the pool once, and both prefect.yaml and SDK deployments get these defaults.
- Docs: Customize base job templates, Manage work pools
B) Deployment via prefect.yaml (production path)
deployments:
- name: my-deployment
entrypoint: flows/my_flow.py:my_flow
work_pool:
name: my-k8s-pool
job_variables:
# Only override if you need to differ from pool defaults
service_account_name: "sa-prod"
annotations:
<http://vpc.amazonaws.com/security-groups|vpc.amazonaws.com/security-groups>: "sg-123,sg-456"
labels:
team: "analytics"
C) Deployment via SDK/internal CLI (playground path) — same keys, same behavior
```
from prefect.flows import FlowMarvin
03/18/2026, 2:27 PMD) Optional: enforce org-wide defaults with Kyverno (applies to all Prefect pods)
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
name: inject-prefect-defaults
spec:
rules:
- name: add-irsa-sgp
match:
resources:
kinds: ["Pod"]
labels:
app.kubernetes.io/managed-by: "prefect"
mutate:
patchStrategicMerge:
metadata:
annotations:
vpc.amazonaws.com/security-groups: "sg-default-a,sg-default-b"
spec:
serviceAccountName: "prefect-default"
Operational tips
- Keep one (or a small number of) Kubernetes work pools with robust templates. Avoid per-team pools; use deployment/job_variables, queues, or admission policies to vary behavior.
- Version-control your base job template JSON and update via CI:
- Get a starting point:
prefect work-pool get-default-base-job-template --type kubernetes > k8s-base.json
- Apply changes:
prefect work-pool update my-k8s-pool --base-job-template ./k8s-base.json
```
- Precedence is predictable: pool defaults < deployment overrides < run-time overrides. Docs: Customize job variables
A couple of questions to tailor this:
- Do your SA/SGP/labels differ by environment (prod vs playground), by team, or both?
- Are Prefect Variables acceptable for environment-scoped defaults (e.g., default_service_account, default_sg_list), or do you prefer admission controllers?
- Do you want your internal CLI to enforce a specific work pool and inject a minimal set of job_variables, or rely entirely on the pool template defaults?
If you share those constraints, I can draft the exact base job template and a small helper for your internal CLI so both paths stay perfectly in sync.Marvin
03/18/2026, 2:46 PMMarvin
03/18/2026, 2:46 PMThang Le
03/18/2026, 2:47 PMMarvin
03/18/2026, 2:47 PMMarvin
03/18/2026, 2:50 PMttlSecondsAfterFinished: "{{ finished_job_ttl }}".
- The finished_job_ttl variable defaults to null (None), so Jobs are retained indefinitely unless you set it.
How to enable automatic deletion
Pick one of these, depending on how broadly you want it applied:
1) Set a default on the work pool’s base job template (recommended)
- Edit your Kubernetes pool’s base job template to give finished_job_ttl a default (e.g., 3600 seconds):
{
"variables": {
"finished_job_ttl": {
"title": "Finished Job TTL (seconds)",
"type": "integer",
"default": 3600
}
},
"job_configuration": {
"job_manifest": {
"spec": {
"ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
}
}
}
}
- CLI helpers (verified):
prefect work-pool get-default-base-job-template --type kubernetes > k8s-base.json
# edit k8s-base.json to set the default
prefect work-pool update my-k8s-pool --base-job-template ./k8s-base.json
Docs: Customize base job templates, Manage work pools
2) Set it per deployment in prefect.yaml
deployments:
- name: my-deployment
work_pool:
name: my-k8s-pool
job_variables:
finished_job_ttl: 3600
3) Set it per run at invocation time
prefect deployment run my-flow/my-deployment --job-variable finished_job_ttl=3600
Cluster requirements
- Kubernetes deletes Jobs based on ttlSecondsAfterFinished via the TTL controller. Most managed clusters have it enabled by default. See K8s docs: TTL after finished
Notes and tips
- TTL applies after the Job reaches a finished state (Complete/Failed). Kubernetes will then garbage-collect the Job (and its Pods) after the TTL.
- If you want “as soon as it’s done,” set a small TTL like 60.
- If you already have a bunch of old Jobs, you can clean them up manually:
kubectl delete job -l <http://app.kubernetes.io/managed-by=prefect|app.kubernetes.io/managed-by=prefect>
References
- Prefect Kubernetes guide: Run flows on Kubernetes
- Prefect Kubernetes integration: prefect-kubernetes
If you share your pool name, I can suggest the exact JSON patch to your base job template to set this org-wide.Thang Le
03/18/2026, 3:08 PM"""Example: register a Prefect KubernetesJob block with custom labels."""
from __future__ import annotations
from prefect_kubernetes.jobs import KubernetesJob # type: ignore[import-untyped]
def create_kubernetes_job_block(
block_name: str,
labels: dict[str, str],
overwrite: bool = True,
) -> KubernetesJob:
"""Create and save a KubernetesJob block with arbitrary labels.
Labels are applied to both the Job metadata and the Pod template
metadata so they propagate to the pods Prefect spawns.
"""
manifest = {
"apiVersion": "batch/v1",
"kind": "Job",
"metadata": {
"labels": dict(labels),
},
"spec": {
"template": {
"metadata": {
"labels": dict(labels),
},
"spec": {
"containers": [
{
"name": "prefect-job",
"env": [],
}
],
"restartPolicy": "Never",
},
},
"backoffLimit": 0,
},
}
block = KubernetesJob(v1_job=manifest)
block.save(block_name, overwrite=overwrite)
return block
if __name__ == "__main__":
create_kubernetes_job_block(
block_name="thang-test-access",
labels={
"app": "prefect-workers-dev",
"environment": "playground",
"thang-test": "true",
},
)Marvin
03/18/2026, 3:08 PMMarvin
03/18/2026, 3:09 PMlabels, annotations, service_account_name).
- Save one block per environment (e.g., “playground”, “prod”).
- In your Kubernetes work pool’s base job template, reference that block with {{ prefect.blocks.<block-type-slug>.<block-name>... }} to populate the Job and Pod metadata.
Example: define and save a custom block
from typing import Dict, Optional
from pydantic import Field
from prefect.blocks.core import Block
class KubernetesDefaults(Block):
_block_type_name = "Kubernetes Defaults"
labels: Dict[str, str] = Field(default_factory=dict)
annotations: Dict[str, str] = Field(default_factory=dict)
service_account_name: Optional[str] = None
# Save per environment
KubernetesDefaults(
labels={"owner":"platform","env":"playground"},
annotations={"<http://example.com/trace|example.com/trace>":"true"},
service_account_name="sa-playground",
).save("playground", overwrite=True)
Then reference it in your base job template
- Put these in your Kubernetes pool’s base job template so every job inherits them automatically.
{
"job_configuration": {
"job_manifest": {
"apiVersion": "batch/v1",
"kind": "Job",
"metadata": {
"labels": "{{ prefect.blocks.kubernetes-defaults.playground.labels }}",
"annotations": "{{ prefect.blocks.kubernetes-defaults.playground.annotations }}"
},
"spec": {
"template": {
"metadata": {
"labels": "{{ prefect.blocks.kubernetes-defaults.playground.labels }}",
"annotations": "{{ prefect.blocks.kubernetes-defaults.playground.annotations }}"
},
"spec": {
"serviceAccountName": "{{ prefect.blocks.kubernetes-defaults.playground.service_account_name }}"
}
}
}
}
}
}
Notes
- Prefect merges your labels/annotations with its own standard labels; your values won’t overwrite Prefect’s required ones.
- If you want to switch environments without editing the template, make the block reference a variable in the template (e.g., k8s_defaults_block_ref) and set its default to one block for playground, then override via deployment/run for prod.
- For org-wide policy (e.g., IRSA/SGP/required labels) admission controllers like Kyverno/Gatekeeper are still the cleanest enforcement layer.
Docs for deeper reference
- Customize base job templates
- Customize/override job variables
- Run flows on Kubernetes
If you share the exact fields you want to inject, I can sketch a ready-to-use base job template plus the minimal Block definition for your setup.Thang Le
03/18/2026, 3:21 PMMarvin
03/18/2026, 3:21 PMMarvin
03/18/2026, 3:24 PM{{ prefect.blocks.kubernetes-defaults.my-block.labels }}
But the block document name segment (my-block) cannot be templated dynamically (e.g., {{ prefect.blocks.kubernetes-defaults[block_name].labels }} or nested {{ ... {{ var }} ... }}) — the resolver expects a static dot-delimited path.
Good, reusable patterns you can use instead
Option A: Make the block name a normal template variable, then resolve it in a small custom step or worker-side code
- Add a variable to your pool template:
{
"variables": {
"k8s_defaults_block_name": {
"title": "K8s defaults block name",
"type": "string",
"default": "playground"
}
}
}
- Don’t try to embed that variable into a block reference; instead:
- At deploy time (or in your internal CLI), resolve the block data by name using the Prefect client and then pass the labels/annotations into job_variables directly:
# Pseudocode inside your CLI/step
block = await client.read_block_document_by_name(
name=k8s_defaults_block_name,
block_type_slug="kubernetes-defaults"
)
job_variables = {
"labels": block.data.get("labels", {}),
"annotations": block.data.get("annotations", {}),
"service_account_name": block.data.get("service_account_name")
}
# pass job_variables on .deploy(...) or via prefect.yaml overrides
This lets you keep “reusability by name” while staying within supported templating.
Option B: Use a static block reference per environment and switch pools or override once
- In the pool template, reference a specific block:
"labels": "{{ prefect.blocks.kubernetes-defaults.playground.labels }}"
- For prod, either:
- Use a different pool that references prod block, or
- Update the pool’s base job template via CI when promoting environments
Option C: Use Prefect Variables to hold the labels, not the block name
- Store labels/annotations as Prefect Variables (or a Secret block for sensitive values), then reference them directly:
"labels": "{{ prefect.variables.prefect_k8s_labels }}"
"annotations": "{{ prefect.variables.prefect_k8s_annotations }}"
- Your internal CLI can update those variables per environment/team.
Why dynamic prefect.blocks.kubernetes-defaults.<name> doesn’t work
- The block resolver parses a single, static, dot-delimited path (block type, block name, optional attribute path). It doesn’t evaluate nested templates or bracket-substitution for the block name segment. As a result, prefect.blocks.kubernetes-defaults[my_var] will not resolve.
If you share how you plan to choose the name (env/team/tag), I can sketch the minimal custom step or CLI snippet to fetch the block by name and inject the labels/annotations/service account into job_variables so your users get a simple, reusable interface.Thang Le
03/18/2026, 3:29 PMMarvin
03/18/2026, 3:29 PMMarvin
03/18/2026, 3:33 PM{
"variables": {
"labels": {
"type": "object",
"default": {}
},
"annotations": {
"type": "object",
"default": {}
},
"service_account_name": {
"type": "string",
"default": null
}
},
"job_configuration": {
"job_manifest": {
"metadata": {
"labels": "{{ labels }}",
"annotations": "{{ annotations }}"
},
"spec": {
"template": {
"metadata": {
"labels": "{{ labels }}",
"annotations": "{{ annotations }}"
},
"spec": {
"serviceAccountName": "{{ service_account_name }}"
}
}
}
}
}
}
Then in your deployment (or via your internal CLI), set the values with block refs:
work_pool:
name: my-k8s-pool
job_variables:
labels: "{{ prefect.blocks.kubernetes-defaults.playground.labels }}"
annotations: "{{ prefect.blocks.kubernetes-defaults.playground.annotations }}"
service_account_name: "{{ prefect.blocks.kubernetes-defaults.playground.service_account_name }}"
Option 2 — Put the block reference into the variable’s default (pool-wide default)
- Keep job_configuration referencing only declared variables.
- Set the default for those variables to the block reference so every job inherits it by default.
Pool base job template:
{
"variables": {
"labels": {
"type": "object",
"default": "{{ prefect.blocks.kubernetes-defaults.playground.labels }}"
},
"annotations": {
"type": "object",
"default": "{{ prefect.blocks.kubernetes-defaults.playground.annotations }}"
},
"service_account_name": {
"type": "string",
"default": "{{ prefect.blocks.kubernetes-defaults.playground.service_account_name }}"
}
},
"job_configuration": {
"job_manifest": {
"metadata": {
"labels": "{{ labels }}",
"annotations": "{{ annotations }}"
},
"spec": {
"template": {
"metadata": {
"labels": "{{ labels }}",
"annotations": "{{ annotations }}"
},
"spec": {
"serviceAccountName": "{{ service_account_name }}"
}
}
}
}
}
}
Important notes
- Don’t place {{ prefect.blocks... }} directly in job_configuration unless it is the value of a declared variable. Otherwise you’ll get the “undeclared variable(s)” error you saw.
- Block references must be static; you cannot dynamically template the block name (e.g., {{ prefect.blocks.kubernetes-defaults[env].labels }} won’t work). If you need dynamic selection, resolve the block in your internal CLI (using the Orchestration client) and inject the resolved dict into job_variables.
- Each string value that contains a block ref must be exactly the placeholder (no extra text around it).
Docs for reference
- Customize base job templates
- Customize/override job variables
If you paste your current base job template JSON, I can mark up the exact changes to make this pass validation.Nate
03/18/2026, 5:07 PM