The same with

memory

Looks like 256 is an acceptable value

And it’s not working as a number or as a string?

Ok thanks, we'll give this a try!

PR for fix: https://github.com/PrefectHQ/prefect/pull/3423

I found it to be a bit tricky to cleanly attach the environment to the flows; so I wrote an internal deployment library that will crawl the files, gather all the flows (essentially imports each file and pulls out all module attributes of type

prefect.Flow

) and update their environments to the proper

FargateTaskEnvironment

before registering them. Also, I am using the

S3Storage

alongside this. You can of course just set it directly on your flows; I just wasn't a fan of that configuration. I wanted my data engineers to not have to be aware of it.

And attach them to the flow and register

If using a custom docker image for the tasks, you need to specify it in the metadata field too

My FargateTaskEnvironment looks something like:

FargateTaskEnvironment(
        # Task Definition
        family=task_definition_name,
        taskRoleArn=settings.task_role_arn,
        executionRoleArn=settings.execution_role_arn,
        cpu=settings.cpu,
        memory=settings.memory,
        containerDefinitions=[
            {
                "name": "flow-container",
                "image": "image",
                "command": [],
                "environment": [],
                "essential": True,
                "logConfiguration": {
                    "logDriver": "awslogs",
                    "options": {
                        "awslogs-group": settings.awslogs_group,
                        "awslogs-region": settings.awslogs_region,
                        "awslogs-stream-prefix": settings.awslogs_stream_prefix,
                    },
                },
            }
        ],
        networkMode="awsvpc",
        requiresCompatibilities=["FARGATE"],

        # Task Run
        cluster=settings.cluster_name,
        region=aws_settings.region,
        taskDefinition=task_definition_name,
        launch_type="FARGATE",
        networkConfiguration={
            "awsvpcConfiguration": {
                "assignPublicIp": "ENABLED"
                if settings.assign_public_ip
                else "DISABLED",
                "subnets": settings.subnets,
            }
        },
        metadata={"image": image},
    )

After checking your above comment on the ECS deployment of the agent now I get what’s happening. The initial env variables are for the task that pulls the flow config not the flow definition itself. Do you suggest a better deployment model for the fargate agent?

The containerDefinitions in the

FargateTaskEnvironment

get overridden by the agent when constructing the task IIRC

I guess it will only pass its own roles and that might not work for us

The task that pulls down the flow from the API uses the task role from the agent configuration. I'm not sure what to call that launching task 🤷‍♂️

Yea we have a single tenancy requirement which demands each customer to have their isolated environment. Our other option would be to delegate executions to something like AWS batch or Databricks and the prefect task would not have direct access to customer data.

I'm not sure how you could have the roles be dynamically applied in the flows.. Perhaps you can specify a dynamic role that the task should assume within the task code? Having a root role that can only

sts:AssumeRole

that the tasks can use directly; then you setup boto3 with an

AssumeRoleCredentialsProvider

(or some such; I know boto3 doesn't have this natively but other AWS SDKs do; for boto3 https://stackoverflow.com/a/45834847) in the session?