Hi there, I am a bit struggling to run two separate flows on ECS and thought to ask for some help 🙂 I would like to run one flow after another. Both flows are registered on prefect cloud, they have their own task_definition_arn and they both run fine separately. I have tried to register flow that would call these flows like in example ( code in the thread). But seems that ECS agent, doesn‘t take in consideration create_flow_run function run_config parameter and tries register new tasks definitions on ECS (or only one for „parent_flow“), although I would like to use existing task definitions plus our ECS agent doesn‘t have permission to register task definitions, therefore I am getting this error:

An error occurred (AccessDeniedException) when calling the RegisterTaskDefinition operation: User: is not authorized to perform: ecs:RegisterTaskDefinition on resource: * because no identity-based policy allows the ecs:RegisterTaskDefinition action

I have tried also to pass flow_a task_definition_arn to parent_flow, then it runs flow, but gets this error:`in create_flow_run ValueError: Received both

flow_id

and

flow_name

. Only one flow identifier can be passed.` Maybe someone can share the best practices how I could create flow that would run in order these two already registered flows ? Big thanks in advance ! 🙂

Hey @Gintautas Jankus, could you move the code to the thread to keep the main channel cleaner? I think your agent really needs the permissions to register the task definition because a new task definition is needed to apply the new RunConfig

Yes, sorry for long post 🙂

FLOW_STORAGE_S3_BUCKET = S3(bucket="simple_bucket")
PROJECT_NAME = "default"
LABELS = ["ecs"]

FLOW_A_RUN_CONFIG = ECSRun(
    task_definition_arn="flow_a_arn",
    labels=LABELS,
)
FLOW_B_RUN_CONFIG = ECSRun(
    task_definition_arn="flow_b_arn",
    labels=LABELS,
)


def workflow() -> Flow:
    with Flow("parent_flow", storage=FLOW_STORAGE_S3_BUCKET) as flow:
        flow_a = create_flow_run(
            flow_name="flow_a", project_name=PROJECT_NAME, run_config=FLOW_A_RUN_CONFIG
        )
        wait_for_flow_a = wait_for_flow_run(flow_a, raise_final_state=True)

        flow_b = create_flow_run(
            flow_name="flow_b", project_name=PROJECT_NAME, run_config=FLOW_B_RUN_CONFIG
        )
        wait_for_flow_b = wait_for_flow_run(flow_b, raise_final_state=True)

        flow_b.set_upstream(wait_for_flow_a)
    return flow


if __name__ == "__main__":
    flow = workflow()
    flow.register(project_name=PROJECT_NAME, labels=LABELS)

Yes, most likely these RunConfigs is not necessary, although I thought that I could somehow avoid creating new task definition as those two separate flows already have their task definitions, but probably will not avoid it and will need to give extra permissions. Thank you 🙂