https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> how does Prefect use Kopf?
# ask-marvin
d
@Marvin how does Prefect use Kopf?
m
✅ thought for 163.6 seconds
Short answer: In Prefect 3.x, Kopf powers the Kubernetes “observer” that runs alongside the Kubernetes worker. It watches your Prefect-created Pods/Jobs and turns their lifecycle events into Prefect events (for automations, UI timelines, alerts), and it can mark runs as crashed when Jobs fail. Prefect 2.x does not use Kopf. What it does - Pod event replication: Watches Pods labeled with 
<http://prefect.io/flow-run-id|prefect.io/flow-run-id>
and emits Prefect events (Added/Running/Succeeded/Failed, including evictions) so you can automate on K8s lifecycle signals. - Crash detection: Watches corresponding Jobs; when backoff limits are hit or no active Jobs remain, it marks the flow run as Crashed. - Runs inside the K8s worker: The worker starts a Kopf operator in a background thread and stops it on shutdown. It can run cluster-wide or be namespace-scoped, and supports multiple observers without duplicate events (via deterministic event IDs). How it’s implemented - Kopf handlers:
Copy code
@kopf.on.startup()
  def configure(settings, **_): settings.scanning.disabled = True

  @kopf.on.event("pods", labels={"<http://prefect.io/flow-run-id|prefect.io/flow-run-id>": kopf.PRESENT})
  async def _replicate_pod_event(...): ...

  @kopf.on.event("jobs", labels={"<http://prefect.io/flow-run-id|prefect.io/flow-run-id>": kopf.PRESENT})
  async def _mark_flow_run_as_crashed(...): ...
- The worker starts/stops the observer thread which runs 
kopf.operator(...)
with either 
clusterwide=True
or explicit 
namespaces=[...]
. Configuration (env vars) - 
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ENABLED
(default true) - 
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_REPLICATE_POD_EVENTS
(default true) - 
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_NAMESPACES
(comma-separated) - 
PREFECT_INTEGRATIONS_KUBERNETES_OBSERVER_ADDITIONAL_LABEL_FILTERS
Logging - Prefect ships a Kopf-specific JSON log formatter to avoid serialization issues when JSON logging is enabled. Source and docs - Observer implementation: prefect_kubernetes/observer.py - Worker integration (starts/stops observer): prefect_kubernetes/worker.py - Settings (env var mapping): prefect_kubernetes/settings.py - Kopf JSON logging: prefect_kubernetes/_logging.py - Prefect K8s guide: Kubernetes integration guide - Kopf docs: Kopf documentation If you’re on Prefect 2.x, Kopf isn’t used; these observer features are 3.x-only.
upvote 1
d
@Marvin I made some changes and now my Prefect 3.4.20 setup on Kubernetes can't start any jobs - they're all 'scheduled' to run but no pods or jobs are kicked off
m
✅ thought for 198.3 seconds
Got it — when runs sit in “Scheduled” and nothing ever starts, it almost always means the run isn’t being claimed by a worker. In Prefect 3.x, a worker polls a work pool for scheduled runs and submits them to Kubernetes. If the worker is down, pointed at the wrong pool/queue, or the pool is paused/misconfigured, nothing will be created in K8s. A few quick checks to triage: 1) Is your worker running and connected to the right pool? - In Kubernetes, check your worker Deployment logs:
Copy code
kubectl logs deploy/<your-worker-deployment> -n <namespace> -f
You should see something like “Started worker 'KubernetesWorker' for pool '<pool-name>'” and periodic “polled x runs” messages. - If you run it manually, make sure you start it with the correct pool:
Copy code
prefect worker start -p <your-pool-name>
2) Is the work pool active, of the right type, and being polled? - List and inspect:
Copy code
prefect work-pool ls
  prefect work-pool inspect "<your-pool-name>"
Verify: - Status: not paused (resume with 
prefect work-pool resume "<pool>"
if needed) - Type: Kubernetes - Concurrency limit: not 0 or fully consumed - Work queues: if you’re using queue filters, confirm they match your deployments’ tags/labels 3) Is your deployment assigned to that pool (and queue) you’re polling? - Inspect the deployment:
Copy code
prefect deployment inspect "<flow-name>/<deployment-name>"
Check “Work pool” and “Work queue” values. If they don’t match the worker, update/redeploy so they do. 4) Did your recent changes alter tags/queues/templates? - If you added a work queue with filters or changed deployment tags, the worker may no longer match. Either adjust the filters or start the worker with the specific queue(s):
Copy code
prefect worker start -p "<pool>" -q "<queue-1>" -q "<queue-2>"
5) Does the worker have the Kubernetes integration installed and permissions? - The worker image/env must include the K8s extra:
Copy code
pip install "prefect[kubernetes]"
If it’s missing, the worker typically fails to start as a Kubernetes worker — check logs. - Ensure the worker’s ServiceAccount can create Jobs/Pods in the target namespace (RBAC). Note: missing RBAC usually causes runs to be claimed then fail; if you never get beyond “Scheduled,” it’s a worker/pool/queue issue. 6) API connectivity - Ensure the worker has 
PREFECT_API_URL
and (if using Cloud) 
PREFECT_API_KEY
set, and it’s pointing to the same workspace as your deployments. If the worker can’t reach the API, it won’t claim runs. If you can share a bit more, I can pinpoint it quickly: - What changes did you make right before this started? - Output of:
Copy code
prefect work-pool ls
  prefect work-pool inspect "<pool>"
  prefect deployment inspect "<flow>/<deployment>"
- How are you running the worker (K8s Deployment/Helm/manual)? A short snippet of the worker logs at startup would help. Docs for reference: - Kubernetes guide: Kubernetes integration guide
d
@Marvin I got this:
Copy code
$ kubectl logs deployment.apps/prefect-worker -n ns-team-arte-prefect -f
Found 2 pods, using pod/prefect-worker-8487f4fc79-nsj2n
/usr/local/lib/python3.11/site-packages/prefect/context.py:955: UserWarning: Failed to create the Prefect home directory at /home/prefect/.prefect
  GLOBAL_SETTINGS_CONTEXT: SettingsContext = root_settings_context()
Loaded collection 'prefect_dask'.
Loaded collection 'prefect_redis'.
Loaded collection 'prefect_kubernetes'.
Worker 'KubernetesWorker 1b3db6e2-847d-41da-b42d-15f5b6a8a6a7' started!
m
✅ thought for 24.2 seconds
Thanks for the logs — that shows your Kubernetes worker did start and the K8s integration is loaded. If runs are stuck in “Scheduled” and no Pods/Jobs are created, the most common causes are: worker not polling the right pool/queues, the pool is paused or constrained by concurrency, or the worker isn’t connected to the same API/workspace as your deployments. Let’s narrow it down quickly. Checks to run 1) Confirm the worker is attached to the correct work pool and actively polling - In the worker Deployment, check the container command/args. It should include the pool:
Copy code
prefect worker start -p <your-pool-name>  # optionally -q <queue>
- After “Worker 'KubernetesWorker ...' started!” you should see periodic polling lines. If you don’t, increase logging and re-check: - Set env in the worker: 
PREFECT_LOGGING_LEVEL=DEBUG
- Then watch logs again:
Copy code
kubectl logs deploy/prefect-worker -n ns-team-arte-prefect -f
2) Inspect your work pool and status - From anywhere with CLI auth to the same workspace:
Copy code
prefect work-pool ls
  prefect work-pool inspect "<your-pool-name>"
Verify: - Status is not paused (resume if needed):
Copy code
prefect work-pool resume "<your-pool-name>"
- Type is Kubernetes - Concurrency limit is not 0 and not fully used. If set to 0 by accident:
Copy code
prefect work-pool clear-concurrency-limit "<your-pool-name>"
3) Ensure deployments target that pool (and queue) - Check the deployment that produced the scheduled runs:
Copy code
prefect deployment inspect "<flow-name>/<deployment-name>"
Confirm “Work pool” matches the one your worker is polling, and if you use work queues, that the names/filters match the worker’s 
-q
flags (or just let the worker poll all queues by omitting 
-q
). 4) Verify API connectivity and workspace alignment - In the worker pod, confirm these env vars:
Copy code
kubectl exec -it deploy/prefect-worker -n ns-team-arte-prefect -- env | grep PREFECT_API
You should see a valid 
PREFECT_API_URL
(and 
PREFECT_API_KEY
if using Cloud). If these are wrong/missing, the worker won’t claim runs from the right workspace. About the warning in your logs - 
Failed to create the Prefect home directory at /home/prefect/.prefect
usually isn’t the root cause here, but it’s worth fixing to avoid oddities with local settings/logging: - Either make 
/home/prefect
writable for the container user, or set
Copy code
PREFECT_HOME=/tmp/.prefect
in the worker env. If you can, please share: - The command/args your worker container uses to start (from the Deployment manifest) - Output of:
Copy code
prefect work-pool inspect "<your-pool-name>"
  prefect deployment inspect "<flow>/<deployment>"
- A few lines of the worker logs after enabling 
PREFECT_LOGGING_LEVEL=DEBUG
Reference: - Kubernetes worker guide: Kubernetes integration guide
d
@Marvin I'm running Prefect 3.4.20 on Kubernetes but my workers can't find the jobs from the server, in the Prefect UI they're scheduled to run but the worker never finds them
m
✅ thought for 64.9 seconds
Thanks for the detail. In Prefect 3.x, when runs stay in “Scheduled” and a Kubernetes worker never picks them up, it’s almost always one of these: - Worker is polling the wrong work pool or specific queues that don’t match your deployments - Work pool is paused or constrained by concurrency - Worker isn’t connected to the same API/workspace as your deployments - Queue filters/tags no longer match after recent changes Let’s run a focused checklist to isolate it. 1) Verify the worker is polling the right pool (and queues) - Check your worker Deployment command/args:
Copy code
kubectl get deploy prefect-worker -n ns-team-arte-prefect -o yaml | grep -A2 -n "prefect worker start"
You should see:
Copy code
prefect worker start -p <your-pool-name>   # optionally -q <queue-1> -q <queue-2>
If you’re passing any 
-q
values, temporarily remove them so the worker polls all queues in the pool. - Enable debug logs and look for polling activity: - Add env var 
PREFECT_LOGGING_LEVEL=DEBUG
to the worker Deployment and restart it. - Then:
Copy code
kubectl logs deploy/prefect-worker -n ns-team-arte-prefect -f
You should see periodic “polled” lines. If you don’t see any polling lines after startup, the worker likely isn’t connected to the right API or pool. 2) Inspect the work pool (status, type, concurrency) - From a machine authenticated to the same workspace:
Copy code
prefect work-pool ls
  prefect work-pool inspect "<your-pool-name>"
Confirm: - Status: not paused (resume if needed)
Copy code
prefect work-pool resume "<your-pool-name>"
- Type: Kubernetes (must match your worker type) - Concurrency limit: not 0 and not fully consumed
Copy code
prefect work-pool clear-concurrency-limit "<your-pool-name>"
3) Verify deployments target that pool (and any queues) - Check the deployment that produced the scheduled runs:
Copy code
prefect deployment inspect "<flow-name>/<deployment-name>"
Ensure: - Work pool matches the one your worker is polling - If you use queues/filters, they match any 
-q
you passed to the worker (or remove 
-q
to poll all) 4) Confirm API connectivity and workspace alignment from the worker pod - Check env:
Copy code
kubectl exec -it deploy/prefect-worker -n ns-team-arte-prefect -- env | grep PREFECT_API
You should see a valid 
PREFECT_API_URL
(and 
PREFECT_API_KEY
if using Cloud). - For Prefect Server: URL typically ends with 
/api
(e.g., http://prefect-api.<ns>.svc.cluster.local:4200/api) - For Prefect Cloud: full account/workspace URL - Quick health check from the worker pod:
Copy code
kubectl exec -it deploy/prefect-worker -n ns-team-arte-prefect -- sh -lc 'apk add --no-cache curl || true; \
    curl -fsS $PREFECT_API_URL/health || curl -fsS $PREFECT_API_URL/api/health || true'
You should get a healthy response. If this fails, the worker can’t reach the API and will never claim runs. 5) Sanity checks - Version alignment:
Copy code
kubectl exec -it deploy/prefect-worker -n ns-team-arte-prefect -- prefect version
Confirm it’s 3.4.20 and matches what your server expects. - UI Work Pools page: open your pool, check “Workers” tab and “Last polled” timestamp. If it never updates, the worker isn’t polling the pool you think it is. Note on the log warning - 
Failed to create the Prefect home directory at /home/prefect/.prefect
is usually harmless but you can eliminate it by setting:
Copy code
PREFECT_HOME=/tmp/.prefect
in the worker env or making 
/home/prefect
writable. It won’t prevent claiming runs. If you can share the following, I can pinpoint the issue quickly: - The worker container command/args from the Deployment - Output of:
Copy code
prefect work-pool inspect "<your-pool-name>"
  prefect deployment inspect "<flow-name>/<deployment-name>"
- A few lines of worker logs at DEBUG level after startup (especially any “polling” or “claim” messages) Reference:
- Kubernetes integration guide: Kubernetes integration guide
d
@Marvin here is the prefect work-pool inspect:
Copy code
$ prefect work-pool inspect arte-k8s          
WorkPool(
    id='73bd42d3-4175-4339-82f4-9cb33b56258d',
    created=DateTime(2024, 12, 5, 23, 5, 30, 735310, tzinfo=Timezone('UTC')),
    updated=DateTime(2025, 11, 20, 22, 50, 2, 188157, tzinfo=Timezone('UTC')),
    name='arte-k8s',
    type='kubernetes',
    base_job_template={
        'variables': {
            'type': 'object',
            'properties': {
                'env': {
                    'type': 'object',
                    'title': 'Environment Variables',
                    'default': [
                        {'name': 'ARTE_REST_TOKEN', 'valueFrom': {'secretKeyRef': {'key': 'arte_rest_token', 'name': 'api-tokens'}}},
                        {'name': 'AUTOMATION_VAULT_TOKEN', 'valueFrom': {'secretKeyRef': {'key': 'arte_vault_token', 'name': 'api-tokens'}}},
                        {'name': 'ARTE__SLACK_BOT_TOKEN', 'valueFrom': {'secretKeyRef': {'key': 'slack_bot_token', 'name': 'api-tokens'}}},
                        {'name': 'ARTE__LOG_LEVEL', 'value': 'INFO'},
                        {'name': 'ARTE_BASE_URL', 'value': '<https://arte.int.ethos12-prod-or1.ethos.adobe.net>'},
                        {'name': 'ARTE__VAULT_URL', 'value': '<https://vault-int-amer.adobe.net>'},
                        {'name': 'ARTE__NAGIOS_USERNAME', 'valueFrom': {'secretKeyRef': {'key': 'nagios_username', 'name': 'nagios'}}},
                        {'name': 'ARTE__NAGIOS_PASSWORD', 'valueFrom': {'secretKeyRef': {'key': 'nagios_password', 'name': 'nagios'}}},
                        {'name': 'USE_SERVICE_TOKEN', 'value': 'true'},
                        {'name': 'ARTE__MYSQL_USERNAME', 'valueFrom': {'secretKeyRef': {'key': 'mysql_username', 'name': 'credentials'}}},
                        {'name': 'ARTE__MYSQL_PASSWORD', 'valueFrom': {'secretKeyRef': {'key': 'mysql_password', 'name': 'credentials'}}},
                        {'name': 'ARTE__FTP_USERNAME', 'valueFrom': {'secretKeyRef': {'key': 'ftp_username', 'name': 'credentials'}}},
                        {'name': 'ARTE__FTP_PASSWORD', 'valueFrom': {'secretKeyRef': {'key': 'ftp_password', 'name': 'credentials'}}},
                        {'name': 'PREFECT_KUBERNETES_CLUSTER_UID', 'value': 'prefect-worker-id-1'},
                        {'name': 'PREFECT_LOGGING_SETTINGS_PATH', 'value': '/etc/prefect/logging/logging.yml'}
                    ],
                    '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': 'prefect-job', 'description': 'Name given to infrastructure created by a worker.'},
                'image': {
                    'anyOf': [{'type': 'string'}, {'type': 'null'}],
                    'title': 'Image',
                    'default': '<http://docker-arte-release.dr-uw2.adobeitc.com/arte-flows:latest|docker-arte-release.dr-uw2.adobeitc.com/arte-flows:latest>',
                    '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',
                    'default': {'app': 'prefect', 'environment': 'production', 'use-default-egress-policy': 'true'},
                    'description': 'Labels applied to infrastructure created by a worker.',
                    'additionalProperties': {'type': 'string'}
                },
                'command': {
                    'anyOf': [{'type': 'string'}, {'type': 'null'}],
                    'title': 'Command',
                    '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': 'ns-team-arte-prefect', 'description': 'The Kubernetes namespace to create jobs within.'},
                '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'}], 'description': 'The Kubernetes cluster config to use for job creation.'},
                'finished_job_ttl': {
                    'anyOf': [{'type': 'integer'}, {'type': 'null'}],
                    'title': 'Finished Job TTL',
                    'default': 300,
                    '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': 'Always', 'description': 'The Kubernetes image pull policy to use for job containers.'},
                'service_account_name': {'anyOf': [{'type': 'string'}, {'type': 'null'}], 'title': 'Service Account Name', 'default': 'prefect-service-account', 'description': 'The Kubernetes service account to use for job creation.'},
                'job_watch_timeout_seconds': {
                    'anyOf': [{'type': 'integer'}, {'type': 'null'}],
                    'title': 'Job Watch Timeout Seconds',
                    '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.'}
            },
            '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.'},
                        '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': {
                            'volumes': [{'name': 'id-rsa', 'secret': {'secretName': 'ssh-key-arte-autobot'}}, {'name': 'config-volume', 'configMap': {'name': 'prefect-logging-config'}}],
                            'containers': [
                                {
                                    'env': '{{ env }}',
                                    'args': '{{ command }}',
                                    'name': 'prefect-job',
                                    'image': '{{ image }}',
                                    'volumeMounts': [
                                        {'name': 'id-rsa', 'subPath': 'ssh-privatekey', 'readOnly': True, 'mountPath': '/home/arte/.ssh/id_rsa', 'mountPropagation': 'None'},
                                        {'name': 'config-volume', 'subPath': 'logging.yml', 'mountPath': '/etc/prefect/logging/logging.yml'}
                                    ],
                                    'imagePullPolicy': '{{ image_pull_policy }}'
                                }
                            ],
                            'completions': 1,
                            'parallelism': 1,
                            'restartPolicy': 'Never',
                            'imagePullSecrets': [{'name': 'artedocker'}],
                            'serviceAccountName': '{{ service_account_name }}'
                        },
                        'metadata': {'labels': '{{ labels }}'}
                    },
                    'backoffLimit': 0,
                    'ttlSecondsAfterFinished': '{{ finished_job_ttl }}'
                },
                'metadata': {'labels': '{{ labels }}', 'namespace': '{{ namespace }}', 'annotations': {'<http://arc.ethos.adobe.net/ignore|arc.ethos.adobe.net/ignore>': 'true', '<http://karpenter.sh/do-not-disrupt|karpenter.sh/do-not-disrupt>': 'true'}, '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 }}'
        }
    },
    status=WorkPoolStatus.READY,
    storage_configuration=WorkPoolStorageConfiguration(),
    default_queue_id='b6aa8565-5598-45ae-85c1-60a7b6327552'
)
m
Your message was too long, here's your message at the allowed limit: here is the prefect work-pool inspect ```$ prefect work-pool inspect arte-k8s WorkPool( id='73bd42d3-4175-4339-82f4-9cb33b56258d' created=DateTime(2024 12 5 23 5 30 735310 tzinfo=Timezone('UTC')) updated=DateTime(2025 11 20 22 50 2 188157 tzinfo=Timezone('UTC')) name='arte-k8s' type='kubernetes' base_job_template={ 'variables' { 'type' 'object' 'properties' { 'env' { 'type' 'object' 'title' 'Environment Variables' 'default' [ {'name' 'ARTE_REST_TOKEN' 'valueFrom' {'secretKeyRef' {'key' 'arte_rest_token' 'name' 'api-tokens'}}} {'name' 'AUTOMATION_VAULT_TOKEN' 'valueFrom' {'secretKeyRef' {'key' 'arte_vault_token' 'name' 'api-tokens'}}} {'name' 'ARTE__SLACK_BOT_TOKEN' 'valueFrom' {'secretKeyRef' {'key' 'slack_bot_token' 'name' 'api-tokens'}}} {'name' 'ARTE__LOG_LEVEL' 'value' 'INFO'} {'name' 'ARTE_BASE_URL' 'value' '<https //arte int ethos12-prod-or1 ethos adobe net>'} {'name' 'ARTE__VAULT_URL' 'value' '<https //vault-int-amer adobe net>'} {'name' 'ARTE__NAGIOS_USERNAME' 'valueFrom' {'secretKeyRef' {'key' 'nagios_username' 'name' 'nagios'}}} {'name' 'ARTE__NAGIOS_PASSWORD' 'valueFrom' {'secretKeyRef' {'key' 'nagios_password' 'name' 'nagios'}}} {'name' 'USE_SERVICE_TOKEN' 'value' 'true'} {'name' 'ARTE__MYSQL_USERNAME' 'valueFrom' {'secretKeyRef' {'key' 'mysql_username' 'name' 'credentials'}}} {'name' 'ARTE__MYSQL_PASSWORD' 'valueFrom' {'secretKeyRef' {'key' 'mysql_password' 'name' 'credentials'}}} {'name' 'ARTE__FTP_USERNAME' 'valueFrom' {'secretKeyRef' {'key' 'ftp_username' 'name' 'credentials'}}} {'name' 'ARTE__FTP_PASSWORD' 'valueFrom' {'secretKeyRef' {'key' 'ftp_password' 'name' 'credentials'}}} {'name' 'PREFECT_KUBERNETES_CLUSTER_UID' 'value' 'prefect-worker-id-1'} {'name' 'PREFECT_LOGGING_SETTINGS_PATH' 'value' '/etc/prefect/logging/logging yml'} ] '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' 'prefect-job' 'description' 'Name given to infrastructure created by a worker '} 'image' { 'anyOf' [{'type' 'string'} {'type' 'null'}] 'title' 'Image' 'default' '<http //docker-arte-release dr-uw2 adobeitc com/arte-flows latest|docker-arte-release dr-uw2 adobeitc com/arte-flows latest>' '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' 'default' {'app' 'prefect' 'environment' 'production' 'use-default-egress-policy' 'true'} 'description' 'Labels applied to infrastructure created by a worker ' 'additionalProperties' {'type' 'string'} } 'command' { 'anyOf' [{'type' 'string'} {'type' 'null'}] 'title' 'Command' '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' 'ns-team-arte-prefect' 'description' 'The Kubernetes namespace to create jobs within '} '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'}] 'description' 'The Kubernetes cluster config to use for job creation '} 'finished_job_ttl' { 'anyOf' [{'type' 'integer'} {'type' 'null'}] 'title' 'Finished Job TTL' 'default' 300 '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' 'Always' 'description' 'The Kubernet
es image pull policy to use for job containers '} 'service_account_name' {'anyOf' [{'type' 'string'} {'type' 'null'}] 'title' 'Service Account Name' 'default' 'prefect-service-account' 'description' 'The Kubernetes service account to use for job creation '} 'job_watch_timeout_seconds' { 'anyOf' [{'type' 'integer'} {'type' 'null'}] 'title' 'Job Watch Timeout Seconds' '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
d
@Marvin is seems like my server and workers are connected but the worker won't pick up new jobs - here is the log from the prefect server:
Copy code
INFO:     172.16.62.228:12862 - "POST /api/logs/filter HTTP/1.1" 200 OK
INFO:     172.17.148.181:40400 - "GET /api/work_pools/arte-k8s HTTP/1.1" 200 OK
INFO:     172.17.148.181:40400 - "POST /api/work_pools/arte-k8s/workers/heartbeat HTTP/1.1" 204 No Content
INFO:     172.16.215.80:60248 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.17.148.181:40400 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "GET /api/flow_runs/8e798cac-ff86-4a28-a3fc-c3e7f39a171b HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "POST /api/logs/filter HTTP/1.1" 200 OK
INFO:     172.16.215.80:60256 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.17.148.181:48618 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "POST /api/flow_runs/paginate HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "GET /api/flow_runs/8e798cac-ff86-4a28-a3fc-c3e7f39a171b HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "POST /api/logs/filter HTTP/1.1" 200 OK
INFO:     172.16.215.80:39640 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.17.148.181:48618 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.16.215.80:39640 - "GET /api/work_pools/arte-k8s HTTP/1.1" 200 OK
INFO:     172.16.215.80:39640 - "POST /api/work_pools/arte-k8s/workers/heartbeat HTTP/1.1" 204 No Content
INFO:     172.16.62.228:12862 - "GET /api/flow_runs/8e798cac-ff86-4a28-a3fc-c3e7f39a171b HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "POST /api/logs/filter HTTP/1.1" 200 OK
INFO:     172.16.215.80:39640 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.17.148.181:48618 - "POST /api/work_pools/arte-k8s/get_scheduled_flow_runs HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "GET /api/flow_runs/8e798cac-ff86-4a28-a3fc-c3e7f39a171b HTTP/1.1" 200 OK
INFO:     172.16.62.228:12862 - "POST /api/logs/filter HTTP/1.1" 200 OK
m
✅ thought for 305.2 seconds
Thanks — those server logs are helpful. They show your Kubernetes worker is heartbeating to the pool and polling it: - GET /api/work_pools/arte-k8s → 200 - POST /api/work_pools/arte-k8s/workers/heartbeat → 204 - POST /api/work_pools/arte-k8s/get_scheduled_flow_runs → 200 So connectivity is fine. The usual reason nothing gets picked up after a successful poll is that the server is returning “no eligible runs” to that worker. That typically happens due to one of: - Work queue mismatch (filters/tags or queue name don’t match the deployment) - Queue or pool paused - Concurrency limits (pool, queue, or deployment) are 0 or fully consumed - Runs are scheduled for the future (time skew / timezone) - Worker restricted to specific queues with -q that don’t include the deployment’s queue - Pool type/worker type mismatch (e.g., Process worker polling a Kubernetes pool) Let’s verify which of these is happening. From a machine authenticated to the same workspace, please run: 1) Find the deployment’s pool and queue - In the UI, open the affected deployment and note “Work pool” and “Work queue”. - Or via CLI:
Copy code
prefect deployment inspect "<flow-name>/<deployment-name>"
2) Check the pool and queue states - Pool status and concurrency:
Copy code
prefect work-pool inspect "arte-k8s"
Ensure it’s not paused and concurrency limit is None or > 0. - List queues and see if any are paused:
Copy code
prefect work-queue ls
In this output, (**) denotes paused queues. If your deployment’s queue is paused:
Copy code
prefect work-queue resume "<queue-name>"
- Preview and artificially poll the exact queue your deployment uses:
Copy code
prefect work-queue preview -p "arte-k8s" "<queue-name>"
  prefect work-queue read-runs -p "arte-k8s" "<queue-name>"
If these return 0 runs while the UI shows Scheduled, it usually means filters/tags or concurrency are blocking eligibility. 3) Check deployment concurrency and scheduled time - For a specific scheduled run ID from the UI:
Copy code
prefect flow-run inspect <flow-run-id>
Look for: - scheduled_time vs current time (from inside the worker pod, 
date -u
). If scheduled_time is in the future due to TZ/time skew, it won’t be picked yet. - deployment_concurrency_limit and current occupancy 4) Confirm the worker is polling the right pool/queues - Show the worker start command/args from your Deployment:
Copy code
kubectl get deploy prefect-worker -n ns-team-arte-prefect -o yaml | grep -nA2 "prefect worker start"
If you pass any 
-q
flags, the worker will ignore other queues in the pool. Remove 
-q
temporarily so it polls all queues in “arte-k8s”. 5) Turn on DEBUG logs for the worker and look for polling details - Set env on worker: 
PREFECT_LOGGING_LEVEL=DEBUG
- Then:
Copy code
kubectl logs deploy/prefect-worker -n ns-team-arte-prefect -f
You should see lines like “Polling work pool arte-k8s” followed by “Found 0 eligible runs” or “Claiming N runs…”. That message tells us if the server is returning empties. If you paste back: - The worker start command/args - 
prefect work-pool inspect "arte-k8s"
- 
prefect work-queue ls
and 
prefect work-queue preview/read-runs -p "arte-k8s" "<queue>"
- 
prefect deployment inspect "<flow>/<deployment>"
(or the flow-run inspect for one scheduled run) …I can pinpoint exactly why get_scheduled_flow_runs is returning nothing for the worker and get you unblocked fast. Reference: Kubernetes integration guide
d
@Marvin is it ok that my work_queue_id is 
None
?
Copy code
$ prefect deployment inspect 'UPGRADE_OS/upgrade_os' | grep -i work
    'description': "Prefect flow/deployment to upgrade the OS of qualified devices.\nCan work by providing a list of hosts or by matching a project/reservation or based on the kernel of 'On Duty - Standby' servers.\nNote, the reservation will be removed 
    'work_queue_name': 'arte-concurrency-1',
    'work_queue_id': None,
    'work_pool_name': 'arte-k8s',
m
✅ thought for 213.8 seconds
Short answer: No — if work_queue_id is None, that deployment’s runs won’t be eligible for any worker. Prefect matches runs to queues by ID, not by the name string in the deployment. What this means - work_queue_name is informational. The server’s scheduling and the worker’s polling use work_queue_id to match runs to a queue. - If the queue object was deleted or never created, deployments can show work_queue_name but have work_queue_id=None. In that state, scheduled runs won’t be returned to workers. Why you might see this - The work queue was deleted; the FK on the deployment is set to NULL (on delete set null). - A legacy/orphaned deployment was created/updated in a way that didn’t set the queue ID. - Recent changes to pools/queues renamed or recreated queues, leaving existing deployments pointing to a non-existent queue. How to fix it 1) Ensure the queue exists in the right pool - In your case: pool “arte-k8s”, queue “arte-concurrency-1” - If it doesn’t exist, create it:
Copy code
prefect work-queue create -p "arte-k8s" "arte-concurrency-1"
2) Re-save the deployment so it binds to the queue ID - If you manage deployments from code, redeploy it so Prefect resolves name → ID:
Copy code
prefect deploy
or re-run your 
flow.from_source(...).deploy(...)
. - Or update it directly:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os"   # confirm fields
  # then update via your deployment config or re-run your deploy command
3) Verify the link is restored - After redeploy/update, check:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os" | grep -i work_queue
You should see both work_queue_name set and work_queue_id populated with a UUID. 4) Make sure your worker is polling that pool/queue - Start worker without -q first (polls all queues in the pool):
Copy code
prefect worker start -p "arte-k8s"
If you do use -q, ensure it includes “arte-concurrency-1”. Optional sanity checks - Preview and artificial poll:
Copy code
prefect work-queue preview -p "arte-k8s" "arte-concurrency-1"
  prefect work-queue read-runs -p "arte-k8s" "arte-concurrency-1"
- Confirm pool/queue not paused and concurrency > 0:
Copy code
prefect work-pool inspect "arte-k8s"
  prefect work-queue ls
If you like, paste the output of: - 
prefect work-queue ls
(look for “arte-concurrency-1” under pool “arte-k8s”) - 
prefect deployment inspect "UPGRADE_OS/upgrade_os"
after redeploy …and I’ll confirm everything is aligned.
d
@Marvin I have PREFECT_LOGGING_LEVEL=DEBUG enabled but I don't see logs with lines like “Polling work pool arte-k8s” followed by “Found 0 eligible runs” or “Claiming N runs…”. There are no more logs after the worker starts:
Copy code
$ kubectl -n ns-team-arte-prefect logs pod/prefect-worker-6864b554f4-hftn2 -f
/usr/local/lib/python3.11/site-packages/prefect/context.py:955: UserWarning: Failed to create the Prefect home directory at /home/prefect/.prefect
  GLOBAL_SETTINGS_CONTEXT: SettingsContext = root_settings_context()
Loaded collection 'prefect_dask'.
Loaded collection 'prefect_redis'.
Loaded collection 'prefect_kubernetes'.
Worker 'KubernetesWorker df328cff-5e0a-478e-999a-9a3beaceb0aa' started!
^C
m
✅ thought for 68.9 seconds
Thanks — that startup log tells me the worker process launched, but the absence of any polling lines after “Worker ... started!” usually means either: - The worker is polling a pool/queue that doesn’t actually have eligible runs, and the debug logs you expect aren’t surfacing, or - The worker can’t see any eligible runs because the deployment isn’t bound to a real work queue (ID), so the server never returns them. Given your earlier inspect shows:
Copy code
'work_queue_name': 'arte-concurrency-1',
'work_queue_id': None,
'work_pool_name': 'arte-k8s',
this is the blocker. Prefect matches runs to queues by ID (not by the name string). If 
work_queue_id
is None, those runs will never be eligible for any worker. Do this to fix quickly 1) Ensure the queue exists in the pool:
Copy code
prefect work-queue create -p "arte-k8s" "arte-concurrency-1"
2) Re-save the deployment so Prefect resolves the queue name → ID: - If you deploy from code, re-run your deploy step (e.g., 
flow.from_source(...).deploy(...)
or 
prefect deploy
). - Then confirm the link:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os" | grep -i work_queue
# Expect both name and a non-None work_queue_id
3) Sanity check the queue and pool: - Not paused and with usable concurrency:
Copy code
prefect work-queue ls
prefect work-pool inspect "arte-k8s"
- Optional: preview and artificially poll the exact queue:
Copy code
prefect work-queue preview -p "arte-k8s" "arte-concurrency-1"
prefect work-queue read-runs -p "arte-k8s" "arte-concurrency-1"
4) Worker flags - Start the worker without any 
-q
filters first so it polls all queues in the pool:
Copy code
prefect worker start -p "arte-k8s"
- If you do pass 
-q
, make sure it includes exactly 
arte-concurrency-1
. About the missing polling logs - On 3.4.20, the “Polling work pool … Found N eligible runs …” lines are DEBUG-level. Make sure 
PREFECT_LOGGING_LEVEL=DEBUG
is set in the container env before the process starts. You can check from inside the pod:
Copy code
prefect config view | grep LOGGING
- Also fix the home directory warning (it can interfere with settings/logging writes):
Copy code
PREFECT_HOME=/tmp/.prefect
or make 
/home/prefect
writable. Once 
work_queue_id
is populated and the worker is polling the right pool/queue, your Scheduled runs should be returned by the server and claimed by the worker. If you still don’t see claims after the steps above, share: - 
prefect work-queue ls
- 
prefect work-pool inspect "arte-k8s"
- 
prefect deployment inspect "UPGRADE_OS/upgrade_os"
- The worker container command/args from the Deployment …and I’ll zero in on the next issue.
d
@Marvin the Prefect worker is throwing this message:
Copy code
20:25:57.113 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Running call get(timeout=1.9999785100808367) in thread 'APILogWorkerThread'
20:25:57.113 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' RUNNING, runtime=0.00> entered
20:25:59.113 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - <WatcherThreadCancelScope, name='get' COMPLETED, runtime=2.00> exited
20:25:59.114 | DEBUG   | APILogWorkerThread | prefect._internal.concurrency - Encountered exception in call get(<dropped>)
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 399, in _run_sync
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/queue.py", line 179, in get
    raise Empty
_queue.Empty
m
✅ thought for 125.2 seconds
That APILogWorkerThread debug trace is benign. It’s Prefect’s internal log-draining thread timing out on an empty queue (
queue.Empty
). It does not indicate an error and doesn’t affect run polling or job submission. Given your deployment shows 
work_queue_id: None
, that’s the real blocker: Prefect matches eligible runs to workers by work_queue_id, not by the name string. With a null ID, the server will never return those runs to any worker, which explains why you don’t see “Found 0/Claiming N runs…” logs even at DEBUG. Fix steps 1) Ensure the queue exists in your pool:
Copy code
prefect work-queue create -p "arte-k8s" "arte-concurrency-1"
2) Re-save/redeploy so the deployment binds to the queue ID: - If you deploy from code, run your deploy again (e.g., 
flow.from_source(...).deploy(...)
or 
prefect deploy
). - Then confirm:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os" | grep -i work_queue
# Expect both work_queue_name set and work_queue_id populated with a UUID
3) Sanity checks:
Copy code
prefect work-queue ls                          # ensure "arte-concurrency-1" exists and isn’t paused
prefect work-queue preview -p "arte-k8s" "arte-concurrency-1"
prefect work-queue read-runs -p "arte-k8s" "arte-concurrency-1"
prefect work-pool inspect "arte-k8s"          # pool not paused; concurrency OK
4) Worker start: - Poll the pool without -q first:
Copy code
prefect worker start -p "arte-k8s"
- If you do use -q, include exactly 
arte-concurrency-1
. Optional: clear the home warning to avoid any config/logging oddities:
Copy code
PREFECT_HOME=/tmp/.prefect
If after this you still don’t see claims, please share: - 
prefect work-queue ls
- 
prefect work-pool inspect "arte-k8s"
- 
prefect deployment inspect "UPGRADE_OS/upgrade_os"
- The worker container command/args from the Deployment Kubernetes guide for reference: Kubernetes integration guide
d
@Marvin is there a way to manually edit a deployments work_queue_id?
m
✅ thought for 144.5 seconds
Short answer: Not directly. In Prefect 3.x, work_queue_id is read-only and computed by the server. You update a deployment’s queue by setting work_pool_name and/or work_queue_name; the server resolves those names to the correct queue ID. Ways to do it - CLI (recommended): redeploy or update using the pool/queue flags
Copy code
prefect deploy ... --pool "arte-k8s" --work-queue "arte-concurrency-1"
- Programmatic API:
Copy code
from prefect.client.orchestration import get_client
  from prefect.client.schemas.actions import DeploymentUpdate

  async with get_client() as client:
      await client.update_deployment(
          "<deployment-id>",
          DeploymentUpdate(
              work_pool_name="arte-k8s",
              work_queue_name="arte-concurrency-1",
          ),
      )
- After updating, verify the ID is populated:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os" | grep -i work_queue
If the named queue doesn’t exist, create it first, then update:
Copy code
prefect work-queue create -p "arte-k8s" "arte-concurrency-1"
Note: Trying to set work_queue_id directly isn’t supported in the client or REST schemas; the server-side logic always derives it from the provided names.
d
@Marvin I tried deleting my deployment and recreating it but Prefect is not updating deployments with the correct UUID
m
✅ thought for 100.2 seconds
Understood. You can’t set the UUID directly, but if Prefect isn’t filling in work_queue_id when you recreate the deployment, it means the server-side name→ID resolution isn’t succeeding. Here’s how to force Prefect to populate the correct ID and common pitfalls to check. What to do now 1) Make sure the pool and queue exist in the same model Prefect expects - Pool must exist: arte-k8s
Copy code
prefect work-pool inspect "arte-k8s"
- Queue must exist in that pool: arte-concurrency-1 - If you created the queue without a pool (legacy queue), Prefect won’t map it when you specify pool+queue. - Create the queue in the pool explicitly:
Copy code
prefect work-queue create -p "arte-k8s" "arte-concurrency-1"
2) Update the deployment using pool + queue names together - This uses the server path that guarantees a pool-scoped queue and will auto-create the queue if missing in that pool. - CLI redeploy:
Copy code
prefect deploy ... --pool "arte-k8s" --work-queue "arte-concurrency-1"
- Or programmatic:
Copy code
from prefect.client.orchestration import get_client
    from prefect.client.schemas.actions import DeploymentUpdate

    async with get_client() as client:
        await client.update_deployment(
            "<deployment-id>",
            DeploymentUpdate(
                work_pool_name="arte-k8s",
                work_queue_name="arte-concurrency-1",
            ),
        )
- Then verify:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os" | grep -i work_queue
  # Expect work_queue_id to be a UUID
3) If you only set work_queue_name (without pool), Prefect may create a legacy queue - The legacy path uses ensure_work_queue_exists(name) which puts the queue in the “default-agent-pool” if present, or creates a standalone queue. This will not bind to arte-k8s. - Always include both pool and queue for modern setups. 4) Double-check there aren’t duplicate queues with the same name across pools - If you previously created a queue named arte-concurrency-1 outside arte-k8s, you could be inspecting that one instead. Confirm the queue under the target pool:
Copy code
prefect work-queue ls | grep arte-concurrency-1
You should see Pool: arte-k8s for the one you intend to use. 5) Inspect your deployment after recreation - Sometimes recreating via a file or recipe retains an old value. Confirm what actually got applied:
Copy code
prefect deployment inspect "UPGRADE_OS/upgrade_os"
If work_pool_name is set but work_queue_id is still None, the queue likely didn’t resolve in that pool. 6) As a diagnostic, try pool-only to bind to the pool’s default queue - This confirms that the pool is healthy and its default queue ID is resolvable:
Copy code
prefect deploy ... --pool "arte-k8s"
After this, inspect the deployment to see if work_queue_id is set to the pool’s default_queue_id. If this still doesn’t populate the UUID, please share: - 
prefect work-pool inspect "arte-k8s"
(look for default_queue_id) - 
prefect work-queue ls
(to confirm a queue named “arte-concurrency-1” exists under the arte-k8s pool) - The exact deploy command or DeploymentUpdate call you’re using - 
prefect deployment inspect "UPGRADE_OS/upgrade_os"
after the update I can then pinpoint whether it’s using the legacy queue path or failing the name→ID lookup.
25 Views
Open in Slack
PreviousNext
Powered by