the flow was created with

prefect

0.12.0. The

KubernetesJobEnvironment

looks like this:

env = KubernetesJobEnvironment(
    job_spec_file="prefect-flow-run.yaml",
    metadata={
        "image": "prefecthq/prefect:all_extras-0.12.0"
    }
)

and

prefect-flow-run.yaml

is exactly the content from https://docs.prefect.io/orchestration/execution/k8s_job_environment.html#job-spec-configuration

I should also add that I built my agent image from the repo as of like two hours ago, to get the newest changes

# peg to a specific state of the Prefect repo
COMMIT_TO_USE="4ac1c8c14e0e85c470e552f5ebab3e07a5891ff4"
REGISTRY_URL="localhost:32000"
IMAGE_TAG="localhost:32000/prefect/prefect-test:${COMMIT_TO_USE:0:6}"

PREFECT_INSTALL_DIR=$(pwd)/prefect-temp
mkdir -p ${PREFECT_INSTALL_DIR}

pushd ${PREFECT_INSTALL_DIR}

    git clone git@github.com:PrefectHQ/prefect.git
    pushd prefect
        git checkout ${COMMIT_TO_USE}

        docker build \
            --build-arg PYTHON_VERSION=3.7 \
            --build-arg GIT_SHA=${COMMIT_TO_USE} \
            --build-arg BUILD_DATE="today" \
            --build-arg PREFECT_VERSION=${COMMIT_TO_USE} \
            --tag ${IMAGE_TAG} \
            -f Dockerfile \
            .

        docker push ${IMAGE_TAG}

    popd
popd

I'd like to come back to this thread. I've gone on a long road through other approaches using

Docker

storage but I'm back to trying

S3

storage today. I understand a lot more about Prefect now than I did when I opened this thread originally, and I think I can ask a more precise question. I'm using this combination • using

flow.register()

with Prefect Cloud • storage:

S3

• agent:

KubernetesAgent

• environment:

KubernetesJobEnvironment

• executor:

LocalExecutor

I understand now that triggering a flow run will kick off this sequence of events: 1. Agent finds a new flow run, gets details from Prefect Cloud 2. Agent creates a k8s job ("Job 1") which run deserialize the storage and run

storage.get_flow()

to get the flow. 3. Job 1 inspects that flow it just got and decides which environment to run it in based on

flow.environment

. In my case, this will create another k8s job ("Job 2") using the job spec I defined in

KubernetesJobEnvironment

, with some other stuff overridden / added (https://github.com/PrefectHQ/prefect/blob/e9da231c1fdc94988b66dd9b336fc98316ecc097/src/prefect/environments/execution/k8s/job.py#L155) 4. Job 2 then runs the tasks in the flow using the executor I provided. In this case I used

LocalExecutor

Here's the issue I'm facing....I cannot figure out how to get AWS creds into "Job 1", and it is failing with the same NoCredentialsError I mentioned above. I see this override option for

IMAGE_PULL_SECRETS

(https://github.com/PrefectHQ/prefect/blob/e9da231c1fdc94988b66dd9b336fc98316ecc097/src/prefect/agent/kubernetes/agent.py#L166) but I can't figure out how to mount in any other secrets. The other suggestions in this thread seem to be about "Job 2" Am I right that this is a gap that's a result of the fact that until very recently the

KubernetesAgent

always just used

Docker

storage?

Thinking about it a little more...I think "all flows from an agent share the same secrets" is ok, since you can make those credentials narrowly scope to pushing and pulling flows. And since

KubernetesJobEnvironment

allows total customization, so people can mount in their own secrets with their own access to databases and things for "Job 2". Maybe places that want total isolation all the way through coud have admins in their Prefect Cloud tenant create one agent per user, generate a UUID as a label, and then give the user that label and say "stick this on your flows and keep it secret". Then only the tenant admin would ever have a view of all agents, and you could guarantee one agent per user.

This would require a bit of work though, so in the immediate future I think we can just make the base template used by the

KubernetesAgent

configurable.