Hi there, I am a bit struggling to run two separat...
# prefect-community
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
. 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
I think there is an existing one that has to be updated. (Not 100% sure)
If you already registered these flows, do you need to pass the RunConfig here? I think you shouldn’t need it since your RunConfig isn’t doing much? Or is it because you need the labels?
Yes, sorry for long post 🙂
Copy code
FLOW_STORAGE_S3_BUCKET = S3(bucket="simple_bucket")
PROJECT_NAME = "default"
LABELS = ["ecs"]


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)

    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 🙂