I have a basic misunderstanding about how the KubernetesJobEnvironment creates jobs. I keep trying to add items into my job_spec_file which gets passed into KubernetesJobEnvironment, but it doesn't seem like they're being picked up. Like, when I try to add spec.template.spec.serviceAccount and spec.template.spec.serviceAccountName, the containers that get create don't mount the token for the specified service account (they just use the

default

one instead). Also, more fundamentally, the prefect code is hard-coded to use the following string for the container args: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/environments/execution/k8s/job.py#L276

"python -c 'import prefect; prefect.Flow.load(prefect.context.flow_file_path).environment.run_flow()'"

but actually the job spec ends up using a different string!

$ kubectl --namespace=prefect get job prefect-job-e4b101a5 -o yaml | grep -A1 args
      - args:
        - prefect execute cloud-flow

I checked the version of prefect on my laptop (which i use to deploy the prefect flow) and the prefect agent image deployed to my cluster. It's all 0.10.x.

Hey @Troy Sankey so a bit of clarification here: the 

prefect-job-xx

 that you’re looking at is the initial prefect job that the agent creates and not the job created by the K8sJobEnvironment. When the Agent finds a flow run it creates a job using the image you have set as your Flow’s 

Docker

 storage. That job acts as a sort of init container where your Flow is loaded, inspected, and executed. In this case it is looking at your flow’s K8sJobEnvironment and creating a new job to execute your flow on. Then the original prefect job completes and the new job that was created runs the flow.

i'll look again and see if i can find it

Do you see anything in the logs of the prefect jobs that could indicate failure?

yea, it clearly was running my code, since it threw an exception on a line in the middle of my task

Just checking, are you attaching the KubernetesJobEnvironment to your

flow.environment

?

yep yep, specified during constructing the Flow. would it help if i pasted this whole flow file?

Yeah any reproduceable example would be helpful 😄

woah, while i was prepping the Gist, i think i noticed a typo:

with Flow("Test credentials", environment=environment, storage=storage) as flow:
    sf_credentials = VaultKVSecret(path="snowflake_pipeline_etl_loader", version=1,)
    test_connection(sf_credentials)

flow.environment = RemoteEnvironment(labels=["prod"])
flow.register(project_name="tsankey test project")

Ah that would do it!

welp, that gives me hope for the next test run 🙂

Yeah currently there isn’t a way to set a default service account for those jobs but there should be! Would you mind opening an issue for that?

sure! i've worked around it by adding a role binding for now.