https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How do we set minimum resources for a...
# ask-marvin
l
@Marvin How do we set minimum resources for a prefect deployment ? I have kubernetes workers, and I need to tell them how much memory my script need to be able to autoscale my kubernetes job when deployment is running... (in Prefect 3x btw)
m
✅ thought for 226.1 seconds
Great question — and you’re on the right track. In Prefect 3.x with Kubernetes workers, you set CPU/memory requests/limits in the Kubernetes work pool’s base job template, then optionally override them per deployment. High-level: - Put resources on the container in the work pool’s base job template (this is what Kubernetes uses to schedule pods and for cluster autoscaling). - Expose them as templated variables so deployments can override per-flow as needed. - Override via prefect.yaml, Python 
.deploy(...)
, or CLI 
prefect deploy --job-variable ...
. Step 1 — add resources to your Kubernetes work pool template - Prefect UI > Work Pools > your K8s pool > Edit > Advanced > Base job template - Add a resources section to the main container and expose variables:
Copy code
{
  "apiVersion": "batch/v1",
  "kind": "Job",
  "metadata": {
    "labels": "{{ labels }}",
    "namespace": "{{ namespace }}",
    "generateName": "{{ name }}-"
  },
  "spec": {
    "backoffLimit": "{{ backoff_limit }}",
    "ttlSecondsAfterFinished": "{{ finished_job_ttl }}",
    "template": {
      "spec": {
        "restartPolicy": "Never",
        "serviceAccountName": "{{ service_account_name }}",
        "containers": [
          {
            "name": "prefect-job",
            "image": "{{ image }}",
            "imagePullPolicy": "{{ image_pull_policy }}",
            "env": "{{ env }}",
            "args": "{{ command }}",
            "resources": {
              "requests": {
                "memory": "{{ memory_request }}",
                "cpu": "{{ cpu_request }}"
              },
              "limits": {
                "memory": "{{ memory_limit }}",
                "cpu": "{{ cpu_limit }}"
              }
            }
          }
        ]
      }
    }
  },
  "variables": {
    "type": "object",
    "properties": {
      "memory_request": {
        "type": "string",
        "default": "256Mi"
      },
      "cpu_request": {
        "type": "string",
        "default": "100m"
      },
      "memory_limit": {
        "type": "string",
        "default": "512Mi"
      },
      "cpu_limit": {
        "type": "string",
        "default": "500m"
      }
    }
  }
}
Step 2 — override per deployment (pick one) - prefect.yaml
Copy code
deployments:
- name: my-deployment
  entrypoint: flows/my_flow.py:my_flow
  work_pool:
    name: my-k8s-pool
    job_variables:
      memory_request: "1Gi"
      cpu_request: "500m"
      memory_limit: "2Gi"
      cpu_limit: "1"
- Python
Copy code
from prefect import flow

@flow
def my_flow():
    ...

my_flow.deploy(
    name="my-deployment",
    work_pool_name="my-k8s-pool",
    job_variables={
        "memory_request": "1Gi",
        "cpu_request": "500m",
        "memory_limit": "2Gi",
        "cpu_limit": "1",
    },
)
- CLI
Copy code
prefect deploy \
  --name my-deployment \
  --pool my-k8s-pool \
  --job-variable memory_request=1Gi \
  --job-variable cpu_request=500m \
  --job-variable memory_limit=2Gi \
  --job-variable cpu_limit=1
Notes - Use standard K8s quantities: memory like 256Mi/1Gi; CPU like 250m/1/2. - Requests determine scheduling and influence cluster autoscaler behavior; limits enforce max usage. For “autoscaling while the deployment is running,” requests are the key inputs for cluster autoscaler to provision nodes that fit your pod. - If every flow run should use the same resources, you can hard-code them in the template instead of templating variables. References - Kubernetes worker docs (base job template and variables): prefect-kubernetes docs - Flow.deploy API (supports job_variables): Prefect docs If you share your current base job template, I can help patch it with the exact resources block you need.
l
@Marvin Okay so, Ive deployed my Prefect instance in my k3s cluster with :
Copy code
apiVersion: <http://argoproj.io/v1alpha1|argoproj.io/v1alpha1>
kind: Application
metadata:
  name: prefect
  namespace: argocd
spec:
  project: default

  sources:
    - repoURL: <https://prefecthq.github.io/prefect-helm>
      chart: prefect-server
      targetRevision: 2025.11.21180339
      helm:
        valuesObject:
          server:
            replicaCount: 1
            
            basicAuth:
              enabled: true
              existingSecret: "prefect-auth-secret"
            
            uiConfig:
              prefectUiApiUrl: "<https://prefect.recomend.app/api>"
            
            env:
              - name: PREFECT_SERVER_API_CSRF_PROTECTION_ENABLED
                value: "true"

          postgresql:
            enabled: true
            primary:
              persistence:
                enabled: true
                storageClass: hcloud-volumes
                size: 10Gi
                annotations:
                  <http://resize.topolvm.io/storage_limit|resize.topolvm.io/storage_limit>: "50Gi"
                  <http://resize.topolvm.io/threshold|resize.topolvm.io/threshold>: "20%"
                  <http://resize.topolvm.io/increase|resize.topolvm.io/increase>: "25%"
                  <http://backup.velero.io/backup-volumes|backup.velero.io/backup-volumes>: "data"

    - repoURL: <https://prefecthq.github.io/prefect-helm>
      chart: prefect-worker
      targetRevision: 2025.11.21180339
      helm:
        valuesObject:
          worker:
            apiConfig: selfHostedServer
            config:
              workPool: kube
              
            selfHostedServerApiConfig:
              apiUrl: "<http://prefect-server.prefect.svc.cluster.local:4200/api>"

              basicAuth:
                enabled: true
                existingSecret: "prefect-auth-secret"

    - repoURL: <https://github.com/recomendapp/infra.git>
      targetRevision: main
      path: apps/services/prefect
      directory:
        recurse: true

  destination:
    server: <https://kubernetes.default.svc>
    namespace: prefect

  syncPolicy:
    automated:
      prune: true
      selfHeal: true
    syncOptions:
      - CreateNamespace=true
      - ServerSideApply=true
I have other file in 
apps/services/prefect
so how Im suppoed to setup cpu memory, etc... ?
m
✅ thought for 206.2 seconds
You’ve got two places to set resources, depending on what you mean by “script”: 1) Worker pod resources (the Prefect worker Deployment) - Configure under the prefect-worker Helm chart values: 
worker.resources
(and optional 
worker.autoscaling
). - This controls the controller pod that submits Jobs, not your flow run pods. 2) Flow run Job resources (the pods that actually execute your flow) - Configure in the Kubernetes work pool’s base job template (and optionally override per deployment via 
job_variables
). - This is what Kubernetes uses for scheduling and cluster autoscaling decisions for each flow run. Below are working snippets tailored to your Argo CD Application (multi-source) setup. A) Set worker pod CPU/memory (prefect-worker chart) Add these to the 
prefect-worker
source’s 
helm.valuesObject
(where you already set 
worker.apiConfig
and 
config.workPool
):
Copy code
worker:
  resources:
    requests:
      cpu: "250m"
      memory: "512Mi"
    limits:
      cpu: "1"
      memory: "2Gi"

  # Optional: HPA for the worker controller pod
  autoscaling:
    enabled: true
    minReplicas: 1
    maxReplicas: 3
    targetCPUUtilizationPercentage: 70
    targetMemoryUtilizationPercentage: 80
B) Set flow run Job CPU/memory (base job template on the work pool) You have two options. Pick one. Option 1 — Inline in Helm values (simple) Add a JSON base job template that sets 
resources
on the main container and exposes variables you can override per deployment:
Copy code
worker:
  config:
    workPool: kube
    baseJobTemplate:
      configuration: |
        {
          "variables": {
            "cpu_request":    { "type": "string", "default": "500m" },
            "memory_request": { "type": "string", "default": "1Gi" },
            "cpu_limit":      { "type": "string", "default": "1" },
            "memory_limit":   { "type": "string", "default": "2Gi" }
          },
          "job_configuration": {
            "job_manifest": {
              "spec": {
                "template": {
                  "spec": {
                    "restartPolicy": "Never",
                    "containers": [
                      {
                        "name": "prefect-job",
                        "resources": {
                          "requests": {
                            "cpu": "{{ cpu_request }}",
                            "memory": "{{ memory_request }}"
                          },
                          "limits": {
                            "cpu": "{{ cpu_limit }}",
                            "memory": "{{ memory_limit }}"
                          }
                        }
                      }
                    ]
                  }
                }
              }
            }
          }
        }
Option 2 — Store JSON in your repo and reference it (nice for GitOps) Since you have 
apps/services/prefect
in your Argo CD app, add a ConfigMap there with the template: apps/services/prefect/base-job-template-cm.yaml ``` apiVersion: v1 kind: ConfigMap metadata: name: prefect-base-job-template namespace: prefect data: baseJobTemplate.json: | { "variables": { "cpu_request": { "type": "string", "default": "500m" }, "memory_request": { "type": "string", "default": "1Gi" }, "cpu_limit": { "type": "string", "default": "1" }, "memory_limit": { "type": "string", "default": "2Gi" } }, "job_configuration": { "job_manifest": { "spec": { "template": { "spec": { "restartPolicy": "Never", "containers": [ { "name": "prefect-job", "resources": { "requests": { "cpu": "{{ cpu_request }}", "memory": "{{ memory_request }}" }, "limits": { "cpu": "{{ cpu_limit }}",
"memory": "{{ memory_limit }}" } } } ] } } } } } }
Copy code
Then point the Helm chart at that ConfigMap in the `prefect-worker` source values:
worker: config: workPool: kube baseJobTemplate: existingConfigMapName: prefect-base-job-template
Copy code
C) Override per deployment (optional, when one flow needs more/less)
- In prefect.yaml:
deployments: - name: heavy entrypoint: flows/heavy.py:run work_pool: name: kube job_variables: cpu_request: "1500m" memory_request: "3Gi" cpu_limit: "2" memory_limit: "5Gi"
Copy code
- Or via CLI:
prefect deploy \ --name heavy \ --pool kube \ --job-variable cpu_request=1500m \ --job-variable memory_request=3Gi \ --job-variable cpu_limit=2 \ --job-variable memory_limit=5Gi ``` Notes - Use standard K8s quantities: memory like 256Mi/1Gi; CPU like 250m/1/2. - Jobs don’t “autoscale” replicas by default; instead, set realistic requests so the cluster autoscaler can provision nodes to fit your pods. If you need horizontal scaling of work, run more concurrent flow runs (or use Dask/Ray/etc. in your flow). - The Helm chart will sync the base job template to the specified work pool on startup; changing these values and letting Argo CD roll the worker is enough to update the pool template. Helpful docs - Prefect Helm charts: prefect-helm - Kubernetes worker docs (base job template + variables): prefect-kubernetes integration - Prefect 3 deployments and job variables: Prefect docs Quick check: Do you want these resources set for the worker pod, the flow run Job pods, or both? I can tailor the exact Argo CD 
Application
snippet accordingly.
l
@Marvin So Ive added in 
apps/services/prefect/
directory :
Copy code
apiVersion: v1
kind: ConfigMap
metadata:
  name: prefect-base-job-template
  namespace: prefect
data:
  baseJobTemplate.json: |
    {
      "variables": {
        "cpu_request":    { "type": "string", "default": "500m" },
        "memory_request": { "type": "string", "default": "1Gi" },
        "cpu_limit":      { "type": "string", "default": "1" },
        "memory_limit":   { "type": "string", "default": "2Gi" }
      },
      "job_configuration": {
        "job_manifest": {
          "spec": {
            "template": {
              "spec": {
                "restartPolicy": "Never",
                "containers": [
                  {
                    "name": "prefect-job",
                    "resources": {
                      "requests": {
                        "cpu": "{{ cpu_request }}",
                        "memory": "{{ memory_request }}"
                      },
                      "limits": {
                        "cpu": "{{ cpu_limit }}",
                        "memory": "{{ memory_limit }}"
                      }
                    }
                  }
                ]
              }
            }
          }
        }
      }
    }
I have others file in 
apps/services/prefect
:
Copy code
─ tree apps/services/prefect ─╯
apps/services/prefect
├── base-job-template-cm.yaml
├── ingress.yaml
├── job-init.yaml
├── kustomization.yaml
└── secrets.yaml
So should I add the new 
base-job-template-cm.yaml
to 
kustomization.yaml
? kustomization.yaml : resources: - ingress.yaml - secrets.yaml - job-init.yaml
m
✅ thought for 57.7 seconds
Short answer: yes — because you’re using Kustomize in that folder, Argo CD will only apply what’s listed in kustomization.yaml. Add your new ConfigMap file there, and also point the prefect-worker chart at it. Do these two things: 1) Add the ConfigMap to kustomization.yaml Update apps/services/prefect/kustomization.yaml so it includes your CM:
Copy code
resources:
- ingress.yaml
- secrets.yaml
- job-init.yaml
- base-job-template-cm.yaml
2) Tell the worker chart to use that ConfigMap In your Argo CD Application, under the prefect-worker source, extend valuesObject to reference the ConfigMap:
Copy code
- repoURL: <https://prefecthq.github.io/prefect-helm>
  chart: prefect-worker
  targetRevision: 2025.11.21180339
  helm:
    valuesObject:
      worker:
        apiConfig: selfHostedServer
        config:
          workPool: kube
          baseJobTemplate:
            existingConfigMapName: prefect-base-job-template
        selfHostedServerApiConfig:
          apiUrl: "<http://prefect-server.prefect.svc.cluster.local:4200/api>"
          basicAuth:
            enabled: true
            existingSecret: "prefect-auth-secret"
After committing, let Argo CD sync. The worker’s init step will sync this base job template into the 
kube
work pool. From then on, each flow run Job will have the CPU/memory requests/limits you templated, and you can still override them per-deployment via 
job_variables
. Sanity checks - Ensure the CM key is exactly 
baseJobTemplate.json
(you have it right). - The CM is created in the same namespace as the worker (
prefect
). - If you don’t see the update take effect, trigger a worker rollout after the CM is applied:
Copy code
kubectl rollout restart deploy -n prefect -l <http://app.kubernetes.io/name=prefect-worker|app.kubernetes.io/name=prefect-worker>
- Verify in Prefect UI → Work Pools → kube → Advanced that the base job template now includes the resources block. Docs for reference: - Helm charts: prefect-helm - K8s worker/job template: prefect-kubernetes integration
l
@Marvin I did all the change but now my worker wont connect to Prefect server :
Copy code
kubectl logs -n prefect deploy/prefect-worker                                                                                                                     ─╯

Defaulted container "prefect-worker" out of: prefect-worker, sync-base-job-template (init)
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/cli/_utilities.py", line 44, in wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/cli/_types.py", line 159, in sync_fn
    return asyncio.run(async_fn(*args, **kwargs))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 654, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/cli/worker.py", line 168, in start
    await worker.start(
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 644, in start
    async with self as worker:
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 997, in __aenter__
    return await super().__aenter__()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 1595, in __aenter__
    await self.setup()
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 936, in setup
    await self.sync_with_backend()
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 1114, in sync_with_backend
    await self._update_local_work_pool_info()
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 1011, in _update_local_work_pool_info
    work_pool = await self._client.create_work_pool(work_pool=wp)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/orchestration/_work_pools/client.py", line 478, in create_work_pool
    response = await self.request(
               ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/orchestration/base.py", line 53, in request
    return await self._client.send(request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 379, in send
    response.raise_for_status()
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 163, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<http://prefect-server.prefect.svc.cluster.local:4200/api/work_pools/>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'type': 'value_error', 'loc': ['body', 'base_job_template'], 'msg': 'Value error, The variables specified in the job configuration template must be present as properties in the variables schema. Your job configuration uses the following undeclared variable(s): cpu_request ,memory_request ,cpu_limit ,memory_limit.', 'input': {'variables': {'cpu_request': {'type': 'string', 'default': '500m'}, 'memory_request': {'type': 'string', 'default': '1Gi'}, 'cpu_limit': {'type': 'string', 'default': '1'}, 'memory_limit': {'type': 'string', 'default': '2Gi'}}, 'job_configuration': {'job_manifest': {'spec': {'template': {'spec': {'restartPolicy': 'Never', 'containers': [{'name': 'prefect-job', 'resources': {'requests': {'cpu': '{{ cpu_request }}', 'memory': '{{ memory_request }}'}, 'limits': {'cpu': '{{ cpu_limit }}', 'memory': '{{ memory_limit }}'}}}]}}}}}}, 'ctx': {'error': {}}}], 'request_body': {'name': 'kube', 'type': 'kubernetes', 'base_job_template': {'variables': {'cpu_request': {'type': 'string', 'default': '500m'}, 'memory_request': {'type': 'string', 'default': '1Gi'}, 'cpu_limit': {'type': 'string', 'default': '1'}, 'memory_limit': {'type': 'string', 'default': '2Gi'}}, 'job_configuration': {'job_manifest': {'spec': {'template': {'spec': {'restartPolicy': 'Never', 'containers': [{'name': 'prefect-job', 'resources': {'requests': {'cpu': '{{ cpu_request }}', 'memory': '{{ memory_request }}'}, 'limits': {'cpu': '{{ cpu_limit }}', 'memory': '{{ memory_limit }}'}}}]}}}}}}}}
For more information check: <https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/422>
An exception occurred.
I guess there is something wrong :
Copy code
apiVersion: v1
kind: ConfigMap
metadata:
  name: prefect-base-job-template
  namespace: prefect
data:
  baseJobTemplate.json: |
    {
      "variables": {
        "cpu_request":    { "type": "string", "default": "500m" },
        "memory_request": { "type": "string", "default": "1Gi" },
        "cpu_limit":      { "type": "string", "default": "1" },
        "memory_limit":   { "type": "string", "default": "2Gi" }
      },
      "job_configuration": {
        "job_manifest": {
          "spec": {
            "template": {
              "spec": {
                "restartPolicy": "Never",
                "containers": [
                  {
                    "name": "prefect-job",
                    "resources": {
                      "requests": {
                        "cpu": "{{ cpu_request }}",
                        "memory": "{{ memory_request }}"
                      },
                      "limits": {
                        "cpu": "{{ cpu_limit }}",
                        "memory": "{{ memory_limit }}"
                      }
                    }
                  }
                ]
              }
            }
          }
        }
      }
    }
We can't merge the default 
baseJobTemplate.json
with my custom one ?
m
Your message was too long, here's your message at the allowed limit: I did all the change but now my worker wont connect to Prefect server ```kubectl logs -n prefect deploy/prefect-worker ─╯ Defaulted container prefect-worker out of prefect-worker sync-base-job-template (init) Traceback (most recent call last) File /usr/local/lib/python3 11/site-packages/prefect/cli/_utilities py line 44 in wrapper return fn(*args *kwargs) ^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/site-packages/prefect/cli/_types py line 159 in sync_fn return asyncio run(async_fn(*args *kwargs)) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/asyncio/runners py line 190 in run return runner run(main) ^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/asyncio/runners py line 118 in run return self _loop run_until_complete(task) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/asyncio/base_events py line 654 in run_until_complete return future result() ^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/site-packages/prefect/cli/worker py line 168 in start await worker start( File /usr/local/lib/python3 11/site-packages/prefect/workers/base py line 644 in start async with self as worker File /usr/local/lib/python3 11/site-packages/prefect_kubernetes/worker py line 997 in aenter return await super() __aenter__() ^^^^^^^^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/site-packages/prefect/workers/base py line 1595 in aenter await self setup() File /usr/local/lib/python3 11/site-packages/prefect/workers/base py line 936 in setup await self sync_with_backend() File /usr/local/lib/python3 11/site-packages/prefect/workers/base py line 1114 in sync_with_backend await self _update_local_work_pool_info() File /usr/local/lib/python3 11/site-packages/prefect/workers/base py line 1011 in _update_local_work_pool_info work_pool = await self _client create_work_pool(work_pool=wp) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/site-packages/prefect/client/orchestration/_work_pools/client py line 478 in create_work_pool response = await self request( ^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/site-packages/prefect/client/orchestration/base py line 53 in request return await self _client send(request) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/site-packages/prefect/client/base py line 379 in send response raise_for_status() File /usr/local/lib/python3 11/site-packages/prefect/client/base py line 163 in raise_for_status raise PrefectHTTPStatusError from_httpx_error(exc) from exc cause prefect exceptions PrefectHTTPStatusError Client error '422 Unprocessable Entity' for url '<http //prefect-server prefect svc cluster local 4200/api/work_pools/>' Response {'exception_message' 'Invalid request received ' 'exception_detail' [{'type' 'value_error' 'loc' ['body' 'base_job_template'] 'msg' 'Value error The variables specified in the job configuration template must be present as properties in the variables schema Your job configuration uses the following undeclared variable(s) cpu_request memory_request cpu_limit memory_limit ' 'input' {'variables' {'cpu_request' {'type' 'string' 'default' '500m'} 'memory_request' {'type' 'string' 'default' '1Gi'} 'cpu_limit' {'type' 'string' 'default' '1'} 'memory_limit' {'type' 'string' 'default' '2Gi'}} 'job_configuration' {'job_manifest' {'spec' {'template' {'spec' {'restartPolicy' 'Never' 'containers' [{'name' 'prefect-job' 'resources' {'requests' {'cpu' '{{ cpu_request }}' 'memory' '{{ memory_request }}'} 'limits' {'cpu' '{{ cpu_limit }}' 'memory' '{{ memory_limit }}'}}}]}}}}}} 'ctx' {'error' {}}}] 'request_body' {'name' 'kube' 'type' 'kubernetes' 'base_job_template' {'variables' {'cpu_request' {'type' 'string' 'default' '500m'} 'memory_request' {'type' 'string' 'default' '1Gi'} 'cpu_limit' {'type' 'string' 'default' '1'} 'memory_limit' {'type' 'string' 'default' '2Gi'}} 'job_configuration' {'job_manifest' {'spec' {'template' {'spec
' {'restartPolicy' 'Never' 'containers' [{'name' 'prefect-job' 'resources' {'requests' {'cpu' '{{ cpu_request }}' 'memory' '{{ memory_request }}'} 'limits' {'cpu' '{{ cpu_limit }}' 'memory' '{{ memory_limit }}'}}}]}}}}}}}} For more information check <https //developer mozilla org/en-US/docs/Web/HTTP/Status/422> An exception occurred ``` I guess there is something wrong ```apiVersion v1 kind ConfigMap metadata name prefect-base-job-template namespace prefect data baseJobTemplate json | { variables { cpu_request { type string default 500m } memory_request { type string default 1Gi } cpu_limit { type string default 1 } memory_limit { type string default 2Gi } } job_configuration { job_manifest { spec { template { spec { restartPolicy Never containers [ { name prefect-job resources { requests { cpu {{ cpu_request }} memory {{ memory_request }} } limits { cpu
l
@Marvin I did all the change but now my worker wont connect to Prefect server :
Copy code
kubectl logs -n prefect deploy/prefect-worker                                                                                                                     ─╯
...
    return await self._client.send(request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 379, in send
    response.raise_for_status()
  File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 163, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<http://prefect-server.prefect.svc.cluster.local:4200/api/work_pools/>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'type': 'value_error', 'loc': ['body', 'base_job_template'], 'msg': 'Value error, The variables specified in the job configuration template must be present as properties in the variables schema. Your job configuration uses the following undeclared variable(s): cpu_request ,memory_request ,cpu_limit ,memory_limit.', 'input': {'variables': {'cpu_request': {'type': 'string', 'default': '500m'}, 'memory_request': {'type': 'string', 'default': '1Gi'}, 'cpu_limit': {'type': 'string', 'default': '1'}, 'memory_limit': {'type': 'string', 'default': '2Gi'}}, 'job_configuration': {'job_manifest': {'spec': {'template': {'spec': {'restartPolicy': 'Never', 'containers': [{'name': 'prefect-job', 'resources': {'requests': {'cpu': '{{ cpu_request }}', 'memory': '{{ memory_request }}'}, 'limits': {'cpu': '{{ cpu_limit }}', 'memory': '{{ memory_limit }}'}}}]}}}}}}, 'ctx': {'error': {}}}], 'request_body': {'name': 'kube', 'type': 'kubernetes', 'base_job_template': {'variables': {'cpu_request': {'type': 'string', 'default': '500m'}, 'memory_request': {'type': 'string', 'default': '1Gi'}, 'cpu_limit': {'type': 'string', 'default': '1'}, 'memory_limit': {'type': 'string', 'default': '2Gi'}}, 'job_configuration': {'job_manifest': {'spec': {'template': {'spec': {'restartPolicy': 'Never', 'containers': [{'name': 'prefect-job', 'resources': {'requests': {'cpu': '{{ cpu_request }}', 'memory': '{{ memory_request }}'}, 'limits': {'cpu': '{{ cpu_limit }}', 'memory': '{{ memory_limit }}'}}}]}}}}}}}}
For more information check: <https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/422>
An exception occurred.
We can't merge the default 
baseJobTemplate.json
with my custom one ?
m
✅ thought for 54.0 seconds
Thanks for the logs — the 422 explains it. Your base job template JSON is close, but the 
variables
section must be a JSON Schema object. Right now you’re listing variables directly; Prefect expects them under 
variables.type
= "object" with 
variables.properties
. Fix your ConfigMap like this:
Copy code
apiVersion: v1
kind: ConfigMap
metadata:
  name: prefect-base-job-template
  namespace: prefect
data:
  baseJobTemplate.json: |
    {
      "variables": {
        "type": "object",
        "properties": {
          "cpu_request":    { "type": "string", "default": "500m" },
          "memory_request": { "type": "string", "default": "1Gi" },
          "cpu_limit":      { "type": "string", "default": "1" },
          "memory_limit":   { "type": "string", "default": "2Gi" }
        }
      },
      "job_configuration": {
        "job_manifest": {
          "spec": {
            "template": {
              "spec": {
                "restartPolicy": "Never",
                "containers": [
                  {
                    "name": "prefect-job",
                    "resources": {
                      "requests": {
                        "cpu": "{{ cpu_request }}",
                        "memory": "{{ memory_request }}"
                      },
                      "limits": {
                        "cpu": "{{ cpu_limit }}",
                        "memory": "{{ memory_limit }}"
                      }
                    }
                  }
                ]
              }
            }
          }
        }
      }
    }
Why the error happened - Prefect validates the template and looks for variables via the JSON Schema in 
variables
. Because the schema wrapper (
type: object
+ 
properties
) was missing, it considered 
cpu_request
, 
memory_request
, etc. as “undeclared,” leading to the 422. About merging with the default template - There’s no automatic merge with the default. The base job template you provide replaces the default entirely. - Recommendation: either - Keep your minimal template like above (fine if you only want to add resources), or - Start from the default Kubernetes template and add the 
resources
block to the main container to retain all the standard placeholders (image, command, env, labels, namespace, etc.). You can view/edit this in the Work Pool UI’s Advanced section or copy from docs/examples. Next steps 1) Update the ConfigMap JSON as above and commit. 2) Ensure it’s included in kustomization.yaml (you already did). 3) Let Argo CD sync. If the worker does not roll automatically, force a rollout:
Copy code
kubectl rollout restart deploy -n prefect -l <http://app.kubernetes.io/name=prefect-worker|app.kubernetes.io/name=prefect-worker>
4) Verify in Prefect UI → Work Pools → your “kube” pool → Advanced that the base job template now shows your resources block. 5) Run a test flow; you can override per-deployment via 
job_variables
as needed. Helpful references - Kubernetes worker and base job template docs: prefect-kubernetes integration - Prefect Helm charts (worker values including baseJobTemplate): prefect-helm If you’d like, I can also give you a “default template + resources added” JSON to drop in, so you keep all the standard fields and just layer in CPU/memory.
l
@Marvin But with this custom 
baseJobTemplate.json
everything is broken. When I run a prefect flow :
Copy code
Worker 'KubernetesWorker b0b8a87a-3574-4925-9b4c-c4b567667609' submitting flow run '5f94d935-4b59-4294-aa34-aef53595d5f6'
06:57:39 PM
prefect.flow_runs.worker
Failed to submit flow run '5f94d935-4b59-4294-aa34-aef53595d5f6' to infrastructure.
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 1284, in _submit_run_and_capture_errors
    configuration = await self._get_configuration(flow_run)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 1395, in _get_configuration
    configuration = await self.job_configuration.from_template_and_values(
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/utilities.py", line 99, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 208, in from_template_and_values
    return cls(**populated_configuration)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/pydantic/main.py", line 250, in __init__
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pydantic_core._pydantic_core.ValidationError: 1 validation error for KubernetesWorkerJobConfiguration
  Value error, Job is missing required attributes at the following paths: /apiVersion, /kind, /spec/template/spec/completions, /spec/template/spec/parallelism [type=value_error, input_value={'job_manifest': {'spec':...artPolicy': 'Never'}}}}}, input_type=dict]
    For further information visit <https://errors.pydantic.dev/2.12/v/value_error>
06:57:39 PM
prefect.flow_runs.worker
Reported flow run '5f94d935-4b59-4294-aa34-aef53595d5f6' as crashed: Flow run could not be submitted to infrastructure:
1 validation error for KubernetesWorkerJobConfiguration
  Value error, Job is missing required attributes at the following paths: /apiVersion, /kind, /spec/template/spec/completions, /spec/template/spec/parallelism [type=value_error, input_value={'job_manifest': {'spec':...artPolicy': 'Never'}}}}}, input_type=dict]
    For further information visit <https://errors.pydantic.dev/2.12/v/value_error>
m
✅ thought for 151.2 seconds
Thanks for the error details — that pinpoints the issue. Your custom template was too minimal; Prefect expects a full Kubernetes Job manifest with specific required fields. The validator is telling us the Job is missing apiVersion, kind, and spec.template.spec.parallelism/completions. Two fixes: Option A — Minimal but valid template (keeps all required defaults and adds resources) Replace the ConfigMap’s JSON with this full template. It preserves Prefect’s defaults and just adds the resources block plus your four variables. ``` apiVersion: v1 kind: ConfigMap metadata: name: prefect-base-job-template namespace: prefect data: baseJobTemplate.json: | { "job_configuration": { "command": "{{ command }}", "env": "{{ env }}", "labels": "{{ labels }}", "name": "{{ name }}", "namespace": "{{ namespace }}", "job_manifest": { "apiVersion": "batch/v1", "kind": "Job", "metadata": { "labels": "{{ labels }}", "namespace": "{{ namespace }}", "generateName": "{{ name }}-" }, "spec": { "backoffLimit": "{{ backoff_limit }}", "ttlSecondsAfterFinished": "{{ finished_job_ttl }}", "template": { "spec": { "parallelism": 1, "completions": 1, "restartPolicy": "Never", "serviceAccountName": "{{ service_account_name }}", "containers": [ { "name": "prefect-job", "env": "{{ env }}", "image": "{{ image }}", "imagePullPolicy": "{{ image_pull_policy }}", "args": "{{ command }}", "resources": { "requests": { "cpu": "{{ cpu_request }}", "memory": "{{ memory_request }}" }, "limits": { "cpu": "{{ cpu_limit }}", "memory": "{{ memory_limit }}" } } } ] } } } }, "cluster_config": "{{ cluster_config }}", "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}", "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}", "stream_output": "{{ stream_output }}" }, "variables": { "type": "object", "properties": { "namespace": { "type": "string", "default": "default", "description": "The Kubernetes namespace to create jobs within." }, "image": { "type": ["string", "null"], "default": null, "description": "The image reference of a container image to use for created jobs." }, "service_account_name": { "type": ["string", "null"], "default": null, "description": "The Kubernetes service account to use for job creation." }, "image_pull_policy": { "type": "string", "enum": ["IfNotPresent", "Always", "Never"], "default": "IfNotPresent", "description": "The Kubernetes image pull policy to use for job containers." }, "backoff_limit": { "type": "integer", "default": 0, "title": "Backoff Limit" }, "finished_job_ttl": { "type": ["integer", "null"], "default": null }, "job_watch_timeout_seconds": { "type": ["integer", "null"], "default": null }, "pod_watch_timeout_seconds": { "type": "integer", "default": 60 }, "stream_output": { "type": "boolean", "default": true },
"cluster_config": { "type": ["object", "null"], "default": null }, "cpu_request": { "type": "string", "default": "500m" }, "memory_request": { "type": "string", "default": "1Gi" }, "cpu_limit": { "type": "string", "default": "1" }, "memory_limit": { "type": "string", "default": "2Gi" } }, "required": [] } }
Copy code
Option B — Start from Prefect’s default and add resources
If you prefer to keep everything the worker ships with, copy the default base job template from your Work Pool UI (Advanced → Base job template → “View JSON”), paste it into the ConfigMap, then add this `resources` block to the main `prefect-job` container and add the four variables to the `variables.properties`:
"resources": { "requests": { "cpu": "{{ cpu_request }}", "memory": "{{ memory_request }}" }, "limits": { "cpu": "{{ cpu_limit }}", "memory": "{{ memory_limit }}" } }
Copy code
After updating
- Commit and let Argo CD sync.
- Restart the worker so it re-syncs the base template:
kubectl rollout restart deploy -n prefect -l app.kubernetes.io/name=prefect-worker ``` - In the UI: Work Pools → your pool → Advanced → confirm the template includes apiVersion/kind/parallelism/completions and your resources. Why the previous error happened - Prefect validates the job manifest against a model that requires: - apiVersion: batch/v1 - kind: Job - spec.template.spec.parallelism (int) - spec.template.spec.completions (int) - restartPolicy: Never - A container named prefect-job with image/args/env placeholders - Your earlier minimal template omitted those, so it failed validation. If you want, paste your current JSON and I’ll patch it to be valid.
l
@Marvin Ive take the default temaplte and merge with the custom one to have :
Copy code
apiVersion: v1
kind: ConfigMap
metadata:
  name: prefect-base-job-template
  namespace: prefect
data:
  baseJobTemplate.json: |
    {
      "variables": {
        "type": "object",
        "properties": {
          "env": {
            "type": "object",
            "title": "Environment Variables",
            "description": "Environment variables to set when starting a flow run.",
            "additionalProperties": {
              "anyOf": [
                {
                  "type": "string"
                },
                {
                  "type": "null"
                }
              ]
            }
          },
          "name": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "title": "Name",
            "default": null,
            "description": "Name given to infrastructure created by a worker."
          },
          "image": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "title": "Image",
            "default": null,
            "examples": [
              "<http://docker.io/prefecthq/prefect:3-latest|docker.io/prefecthq/prefect:3-latest>"
            ],
            "description": "The image reference of a container image to use for created jobs. If not set, the latest Prefect image will be used."
          },
          "labels": {
            "type": "object",
            "title": "Labels",
            "description": "Labels applied to infrastructure created by a worker.",
            "additionalProperties": {
              "type": "string"
            }
          },
          "command": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "title": "Command",
            "default": null,
            "description": "The command to use when starting a flow run. In most cases, this should be left blank and the command will be automatically generated by the worker."
          },
          "namespace": {
            "type": "string",
            "title": "Namespace",
            "default": "prefect",
            "description": "The Kubernetes namespace to create jobs within."
          },
          "backoff_limit": {
            "type": "integer",
            "title": "Backoff Limit",
            "default": 0,
            "minimum": 0,
            "description": "The number of times Kubernetes will retry a job after pod eviction. If set to 0, Prefect will reschedule the flow run when the pod is evicted unless PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR is set to value different from 'reschedule'."
          },
          "stream_output": {
            "type": "boolean",
            "title": "Stream Output",
            "default": true,
            "description": "If set, output will be streamed from the job to local standard output."
          },
          "cluster_config": {
            "anyOf": [
              {
                "$ref": "#/definitions/KubernetesClusterConfig"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "description": "The Kubernetes cluster config to use for job creation."
          },
          "finished_job_ttl": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "title": "Finished Job TTL",
            "default": null,
            "description": "The number of seconds to retain jobs after completion. If set, finished jobs will be cleaned up by Kubernetes after the given delay. If not set, jobs will be retained indefinitely."
          },
          "image_pull_policy": {
            "enum": [
              "IfNotPresent",
              "Always",
              "Never"
            ],
            "type": "string",
            "title": "Image Pull Policy",
            "default": "IfNotPresent",
            "description": "The Kubernetes image pull policy to use for job containers."
          },
          "service_account_name": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "title": "Service Account Name",
            "default": null,
            "description": "The Kubernetes service account to use for job creation."
          },
          "job_watch_timeout_seconds": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "title": "Job Watch Timeout Seconds",
            "default": null,
            "description": "Number of seconds to wait for each event emitted by a job before timing out. If not set, the worker will wait for each event indefinitely."
          },
          "pod_watch_timeout_seconds": {
            "type": "integer",
            "title": "Pod Watch Timeout Seconds",
            "default": 60,
            "description": "Number of seconds to watch for pod creation before timing out."
          },
          "cpu_request": {
            "type": "string",
            "title": "CPU Request",
            "default": "500m",
            "description": "The requested CPU for the job."
          },
          "memory_request": {
            "type": "string",
            "title": "Memory Request",
            "default": "1Gi",
            "description": "The requested memory for the job."
          },
          "cpu_limit": {
            "type": "string",
            "title": "CPU Limit",
            "default": "1",
            "description": "The CPU limit for the job."
          },
          "memory_limit": {
            "type": "string",
            "title": "Memory Limit",
            "default": "2Gi",
            "description": "The memory limit for the job."
          }
        },
        "definitions": {
          "KubernetesClusterConfig": {
            "type": "object",
            "title": "KubernetesClusterConfig",
            "required": [
              "config",
              "context_name"
            ],
            "properties": {
              "config": {
                "type": "object",
                "title": "Config",
                "description": "The entire contents of a kubectl config file.",
                "additionalProperties": true
              },
              "context_name": {
                "type": "string",
                "title": "Context Name",
                "description": "The name of the kubectl context to use."
              }
            },
            "description": "Stores configuration for interaction with Kubernetes clusters.\n\nSee `from_file` for creation.",
            "secret_fields": [],
            "block_type_slug": "kubernetes-cluster-config",
            "block_schema_references": {}
          }
        },
        "description": "Default variables for the Kubernetes worker.\n\nThe schema for this class is used to populate the `variables` section of the default\nbase job template."
      },
      "job_configuration": {
        "env": "{{ env }}",
        "name": "{{ name }}",
        "labels": "{{ labels }}",
        "command": "{{ command }}",
        "namespace": "{{ namespace }}",
        "job_manifest": {
          "kind": "Job",
          "spec": {
            "template": {
              "spec": {
                "containers": [
                  {
                    "env": "{{ env }}",
                    "args": "{{ command }}",
                    "name": "prefect-job",
                    "image": "{{ image }}",
                    "imagePullPolicy": "{{ image_pull_policy }}",
                    "resources": {
                      "requests": {
                        "cpu": "{{ cpu_request }}",
                        "memory": "{{ memory_request }}"
                      },
                      "limits": {
                        "cpu": "{{ cpu_limit }}",
                        "memory": "{{ memory_limit }}"
                      }
                    }
                  }
                ],
                "completions": 1,
                "parallelism": 1,
                "restartPolicy": "Never",
                "serviceAccountName": "{{ service_account_name }}"
              }
            },
            "backoffLimit": "{{ backoff_limit }}",
            "ttlSecondsAfterFinished": "{{ finished_job_ttl }}"
          },
          "metadata": {
            "labels": "{{ labels }}",
            "namespace": "{{ namespace }}",
            "generateName": "{{ name }}-"
          },
          "apiVersion": "batch/v1"
        },
        "stream_output": "{{ stream_output }}",
        "cluster_config": "{{ cluster_config }}",
        "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}",
        "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}"
      }
    }
So I can use it with ?
Copy code
- name: sync_typesense
  version: 1.0.0
  tags:
  - typesense
  description: Sync all Typesense collections (movies, tv_series, persons, playlists, users)
  entrypoint: sync_typesense/sync_typesense.py:sync_typesense
  parameters: {}
  work_pool:
    name: kube
    job_variables:
      <<: *job_variables
      memory_request: "8Gi"
      cpu_request: "2"
      memory_limit: "8Gi"
      cpu_limit: "4"
m
Your message was too long, here's your message at the allowed limit: Ive take the default temaplte and merge with the custom one to have ```apiVersion v1 kind ConfigMap metadata name prefect-base-job-template namespace prefect data baseJobTemplate json | { variables { type object properties { env { type object title Environment Variables description Environment variables to set when starting a flow run additionalProperties { anyOf [ { type string } { type null } ] } } name { anyOf [ { type string } { type null } ] title Name default null description Name given to infrastructure created by a worker } image { anyOf [ { type string } { type null } ] title Image default null examples [ <http //docker io/prefecthq/prefect 3-latest|docker io/prefecthq/prefect 3-latest> ] description The image reference of a container image to use for created jobs If not set the latest Prefect image will be used } labels { type object title Labels description Labels applied to infrastructure created by a worker additionalProperties { type string } } command { anyOf [ { type string } { type null } ] title Command default null description The command to use when starting a flow run In most cases this should be left blank and the command will be automatically generated by the worker } namespace { type string title Namespace default prefect description The Kubernetes namespace to create jobs within } backoff_limit { type integer title Backoff Limit default 0 minimum 0 description The number of times Kubernetes will retry a job after pod eviction If set to 0 Prefect will reschedule the flow run when the pod is evicted unless PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR is set to value different from 'reschedule' } stream_output { type boolean title Stream Output default true description If set output will be streamed from the job to local standard output } cluster_config { anyOf [ { $ref #/definitions/KubernetesClusterConfig } { type null } ] default null description The Kubernetes cluster config to use for job creation } finished_job_ttl { anyOf [ { type integer } { type null } ] title Finished Job TTL default null description The number of seconds to retain jobs after completion If set finished jobs will be cleaned up by Kubernetes after the given delay If not set jobs will be retained indefinitely } image_pull_policy { enum [ IfNotPresent Always Never ] type string title Image Pull Policy default IfNotPresent description The Kubernetes image pull policy to use for job containers } service_account_name { anyOf [ { type string } { type null } ] title Service Account Name default null description The Kubernetes service account to use for job creation } job_watch_timeout_seconds { anyOf [ { type integer } { type null } ] title Job Watch Timeout Seconds default null description Number of seconds to wait for each event emitted by a job before timing out If not set the worker will wait for each event indefinitely } pod_watch_timeout_seconds { type integer title Pod Watch Timeout Seconds default 60 description Number of seconds to watch
10 Views
Open in Slack
PreviousNext
Powered by