<@ULVA73B9P> I am using prefect 2.20 and want to r...
# marvin-ai
f
@Marvin I am using prefect 2.20 and want to run flow runs as k8s jobs. The image required to be pulled within the k8s pod is stored in a custom azure container registry. Can I specify this in the pull step within the
prefect.yaml
file?
m
thought for 335.6 seconds
Short answer: No. In Prefect 2.20 the
pull
step in
prefect.yaml
is only for fetching your flow code (e.g., from Git or remote storage) inside the running container. It does not control container image pulling. Image pulling is handled by Kubernetes via the pod spec. How to run on K8s with a private ACR image instead You have two common paths in Prefect 2.20: 1) Kubernetes work pool (prefect-kubernetes worker) – recommended - Set the image in your deployment’s
work_pool.job_variables
. - To use imagePullSecrets for ACR, add a variable and wire it into the base job template in the work pool’s Advanced settings. Example
prefect.yaml
(after you’ve added
image_pull_secrets
to the pool’s template):
Copy code
deployments:
  - name: k8s
    entrypoint: flows/hello.py:flow
    work_pool:
      name: my-k8s-pool
      job_variables:
        image: <http://myregistry.azurecr.io/my-image:latest|myregistry.azurecr.io/my-image:latest>
        namespace: default
        image_pull_policy: IfNotPresent
        image_pull_secrets:
          - name: acr-secret
Example base job template snippet (in the work pool’s Advanced editor) showing how to add and use `image_pull_secrets`:
Copy code
{
  "variables": {
    "properties": {
      "image": {
        "type": "string"
      },
      "namespace": {
        "type": "string",
        "default": "default"
      },
      "image_pull_policy": {
        "type": "string",
        "enum": [
          "IfNotPresent",
          "Always",
          "Never"
        ],
        "default": "IfNotPresent"
      },
      "image_pull_secrets": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "name": {
              "type": "string"
            }
          }
        },
        "default": []
      }
    }
  },
  "job_configuration": {
    "job_manifest": {
      "apiVersion": "batch/v1",
      "kind": "Job",
      "spec": {
        "template": {
          "spec": {
            "imagePullSecrets": "{{ image_pull_secrets }}",
            "containers": [
              {
                "name": "prefect-job",
                "image": "{{ image }}",
                "imagePullPolicy": "{{ image_pull_policy }}"
              }
            ],
            "restartPolicy": "Never"
          }
        }
      }
    }
  }
}
2) Legacy KubernetesJob infrastructure block - Set the
image
normally, and add
imagePullSecrets
via
customizations
(JSON patch to the pod spec):
Copy code
from prefect.infrastructure.kubernetes import KubernetesJob

k8s = KubernetesJob(
    image="<http://myregistry.azurecr.io/my-image:latest|myregistry.azurecr.io/my-image:latest>",
    namespace="default",
    image_pull_policy="IfNotPresent",
    customizations=[
        {
            "op": "add",
            "path": "/spec/template/spec/imagePullSecrets",
            "value": [{"name": "acr-secret"}]
        }
    ]
)
k8s.save("k8s-job", overwrite=True)
Create the Kubernetes secret for ACR (if not using AKS’s automatic integration):
Copy code
kubectl create secret docker-registry acr-secret \
  --docker-server=<http://myregistry.azurecr.io|myregistry.azurecr.io> \
  --docker-username=<ACR_USERNAME> \
  --docker-password=<ACR_PASSWORD_OR_TOKEN> \
  --namespace=default
Tip for AKS + ACR - If you run AKS, attaching ACR at the cluster level often removes the need to explicitly set `imagePullSecrets`:
Copy code
az aks update -g <resource-group> -n <cluster-name> --attach-acr <acr-name>
- Docs: AKS + ACR integration References - KubernetesJob source (2.20): Kubernetes infrastructure - Pull steps (code fetching, not image pulling): Pull steps - Kubernetes worker (prefect-kubernetes): prefect-kubernetes integration
Would you like guidance editing your work pool’s base job template, or are you using the legacy
KubernetesJob
block right now? Also, are you on AKS with ACR attached, or do you need to reference a specific secret name?
f
@Marvin When using Prefect 2.20, and trying to create work_pools as shown below, even though I have set the overwrite to True, the work pools are not updated after this step. I checked this using the Prefect UI dashboard. What can be the reason for this?
Copy code
flow_name = flow._name_
        flow_specific_work_pool_name = f"{flow_name}"
        async with get_client() as client:
            await client.create_work_pool(
                work_pool=WorkPoolCreate(
                    name=flow_specific_work_pool_name,
                    type=worker.type,
                    base_job_template=worker._base_job_template,
                ),
                overwrite=True
            )
m
thought for 210.4 seconds
Likely causes in Prefect 2.20 - overwrite=True does not change the pool type - If the existing pool’s type differs from the one you pass, Prefect will keep the original type and emit a warning (“Overwriting work pool type is not supported. Ignoring provided type.”). The UI will continue to show the old type. - Only these fields are updated on overwrite: description, is_paused, base_job_template, concurrency_limit. - Source (2.20.0): WorkPoolUpdate fields and create_work_pool logic - UI refresh/caching delay - The UI may cache work pool data briefly; you may need to refresh or wait ~30s to see changes, especially for base_job_template updates. - You might be updating a different workspace/API than the UI - Ensure your code’s PREFECT_API_URL (and API key if using Cloud) matches the workspace you’re viewing in the UI. - Quick check:
Copy code
from prefect.settings import get_current_settings
    print(get_current_settings().api_url)
- No actual change was sent - If
worker._base_job_template
is None or identical to what’s already stored, you won’t see a visible change in the UI. - Confirm what was applied by reading it back after the call:
Copy code
async with get_client() as client:
        updated = await client.read_work_pool(flow_specific_work_pool_name)
        print(updated.base_job_template)
- Silent warning not surfaced - The “cannot overwrite type” warning may be easy to miss in async or non-verbose runs. You can force warnings to display:
Copy code
import warnings
    warnings.simplefilter("default")
What your code does on conflict with overwrite=True - Prefect first tries to create the pool. If it already exists (409), it: - Keeps the existing type (ignores the new type, with a warning) - Updates only: description, is_paused, base_job_template, concurrency_limit - Returns the updated pool - Relevant code (2.20.0): OrionClient.create_work_pool and WorkPoolCreate/WorkPoolUpdate Sanity-check snippet
Copy code
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate, WorkPoolUpdate

name = "my-pool"
wp_create = WorkPoolCreate(
    name=name,
    type="kubernetes",             # Will NOT change if pool already exists with different type
    base_job_template=worker._base_job_template,  # Must be a dict; ensure not None
)

async with get_client() as client:
    res = await client.create_work_pool(work_pool=wp_create, overwrite=True)
    print("Returned type:", res.type)
    # Verify server state
    fetched = await client.read_work_pool(name)
    print("Fetched base_job_template keys:", fetched.base_job_template.keys())
If you need to update only the base_job_template explicitly:
Copy code
async with get_client() as client:
    await client.update_work_pool(
        work_pool_name=name,
        work_pool=WorkPoolUpdate(base_job_template=worker._base_job_template),
    )
Questions to pinpoint the cause - Which field(s) are you expecting to change (type, base_job_template, description, etc.)? - Are you certain your code points at the same Prefect workspace the UI is showing? - Can you print the WorkPool returned by create_work_pool and confirm the values there? - What does
worker._base_job_template
contain (and is it a non-empty dict)?