It's a safety mechanism

The new environment doesn't match the environment that's going to be used by your flows, so it errors out since there is a mismatch and gives you some approaches to address it at the end of the error message.

The error explains it

It says there's an existing task definition that matches the name exactly

The pre-existing task definition has different values than is provided in the new environment. As the FargateTaskEnvironment currently runs, it would have NOT used the new values which is, presumably, not desirable (otherwise, why did you change the values?). Prior to this error, your flows would run but with the OLD values (cpu/memory, etc) and thus not behave as anticipated.

Where are the "expected" values coming from if not from the AWS API? Is there no taskDefinition of family

00-orchestrator

? Perhaps it's in a different region?

And this is how I define the environment for the flow

environment = FargateTaskEnvironment(
    cpu="256",
    memory="1024",
    family=ETL_FLOW_NAME,
    taskDefinition=ETL_FLOW_NAME,
    taskRoleArn=ETL_TASK_ROLE_ARN,
    executionRoleArn=ETL_EXECUTION_ROLE_ARN,
    requiresCompatibilities=["FARGATE"],
    containerDefinitions=[
        {
            "name": "flow",
            "image": ETL_IMAGE_NAME,
            "command": [
                "/bin/sh",
                "-c",
                "prefect execute flow-run"
            ],
            "environment": [],
            "essential": True,
        }
    ]
)

Here's the code: https://github.com/PrefectHQ/prefect/blob/35e489dbabd8f1b6cabbbe822cf9af3b2a5b86f5/src/prefect/environments/execution/fargate/fargate_task.py#L318-L327

It literally tries to access the task definition via boto3, and if it doesn't exist, it creates it

The error you're getting can only happen if the task definition exists prior to the flow run

It creates a separate task definition using the values passed to the Fargate Agent

I already explained this yesterday

The agent runs a task, which in turn runs your flow in another task

The agent will create a taskDefinition for this intermediate task

This intermediate task != your flow's task and as such they do not share the same taskDefinition.

Perhaps you've used the same

family

for your Agent's configuration, as you have for the

FargateTaskEnvironment

? Which would cause this issue

However, it’s not clear where I should set the agent task definition

family

Finally I was able to get a flow running. However, it runs successfully the first time. From the second run it gives me the following error

Failed to load and execute Flow's environment: ValueError("The given taskDefinition does not match the existing taskDefinition 01-process-event-flow.\nDetail: \n\

tcontainerDefinition.0.environment -> 
Given: [
{'name': 'PREFECT__CLOUD__GRAPHQL', 'value': '<https://apollo-prefect-server.srv-stage.cloudacademy.xyz/graphql/graphql>'}, 
{'name': 'PREFECT__CLOUD__USE_LOCAL_SECRETS', 'value': 'false'}, 
{'name': 'PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudFlowRunner'}, 
{'name': 'PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudTaskRunner'}, 
{'name': 'PREFECT__LOGGING__LOG_TO_CLOUD', 'value': 'true'}, 
{'name': 'PREFECT__LOGGING__EXTRA_LOGGERS', 'value': '[]'}
], 

Expected: [
{'name': 'PREFECT__CLOUD__GRAPHQL', 'value': '<https://apollo-prefect-server.srv-stage.cloudacademy.xyz/graphql/graphql>'}
{'name': 'PREFECT__CLOUD__USE_LOCAL_SECRETS', 'value': 'false'}, 
{'name': 'PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudFlowRunner'}, 
{'name': 'PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudTaskRunner'}, 
{'name': 'PREFECT__LOGGING__LOG_TO_CLOUD', 'value': 'true'}, 
{'name': 'PREFECT__LOGGING__EXTRA_LOGGERS', 'value': '[]'}, 

]

\n\nIf the given configuration is desired, deregister the existing\ntaskDefinition and re-run the flow. Alternatively, you can\nchange the family/taskDefinition name in the FargateTaskEnvironment\nfor this flow.")

But I don’t see differences between Given and Expected 😅

Here’s another example

Failed to load and execute Flow's environment: ValueError("The given taskDefinition does not match the existing taskDefinition 01-process-event-flow.\nDetail: \n\tcontainerDefinition.0.environment -> Given: [{'name': 'PREFECT__CLOUD__GRAPHQL', 'value': '<https://apollo-prefect-server.srv-stage.cloudacademy.xyz/graphql/graphql>'}, {'name': 'PREFECT__CLOUD__USE_LOCAL_SECRETS', 'value': 'false'}, {'name': 'PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudFlowRunner'}, {'name': 'PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudTaskRunner'}, {'name': 'PREFECT__LOGGING__LOG_TO_CLOUD', 'value': 'true'}, {'name': 'PREFECT__LOGGING__EXTRA_LOGGERS', 'value': '[]'}], Expected: [{'name': 'PREFECT__CLOUD__USE_LOCAL_SECRETS', 'value': 'false'}, {'name': 'PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudFlowRunner'}, {'name': 'PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS', 'value': 'prefect.engine.cloud.CloudTaskRunner'}, {'name': 'PREFECT__LOGGING__LOG_TO_CLOUD', 'value': 'true'}, {'name': 'PREFECT__LOGGING__EXTRA_LOGGERS', 'value': '[]'}, {'name': 'PREFECT__CLOUD__GRAPHQL', 'value': '<https://apollo-prefect-server.srv-stage.cloudacademy.xyz/graphql/graphql'}>]\n\nIf the given configuration is desired, deregister the existing\ntaskDefinition and re-run the flow. Alternatively, you can\nchange the family/taskDefinition name in the FargateTaskEnvironment\nfor this flow.")

I can confirm that the task definition in AWS it’s the same as the one in

Given

. The problem seems to be the validation between the elements in

environment

which are in different order. @Spencer does this make sense to you?

Should be a straight-forward fix; update https://github.com/PrefectHQ/prefect/blob/f554d1313638286d4183b136f97b56bfbfae58db/src/prefect/environments/execution/fargate/fargate_task.py#L255-L256 with something like:

given=sorted(value) if key == "environment" else value,
  expected=sorted(existing_container_definition.get(key)) if key == "environment" else value,

I just opened this issue https://github.com/PrefectHQ/prefect/issues/3451

actually scratch that.. it’s working fine 😬.. wrong version pinned 🤦