Running into this error when attempting to run a flow using ECS agent and ECSRun:
I have a working config for prefect agent that executes the flow without errors. However, this involves creating a task-definitions.yaml:
botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameterException) when calling the RunTask operation: Task definition does not support launch_type FARGATE.
prefect agent ecs start -t token \ -n aws-ecs-agent \ -l label \ --task-definition /path/to/task-definition.yaml \ --cluster cluster_arn
The flow runs without errors, so the error is not due to IAM permissions. However, when running the ECS Agent using the
networkMode: awsvpc cpu: 1024 memory: 2048 taskRoleArn: task_role_arn executionRoleArn: execution_role_arn
CLI args, I run into the above-mentioned error. I have also tried running Prefect agent using
, which I believe is the default and does not need to be specified, but this does not work too.
I have also tried to pass in
prefect agent ecs start -t token \ -n aws-ecs-agent \ -l ecs \ --task-role-arn task_role_arn \ --execution-role-arn execution_role_arn \ --cluster cluster_arn
into the ECSRun() function within my flow, and ran into the same error. Is there any way to run ECS Agent using CLI args without using the task-definition file?
— I set up all agents/cli😒 two days ago at their latest versions, and the agent is a k8s one. I'm using Prefect cloud. This happens when I register the flow as such and then run it from the UI (I verified the UI archives the old version and I'm triggering a run of the updated version)
Unexpected error while running flow: KeyError('Task slug resolve_profiles_dir-1 not found in the current Flow; this is usually caused by changing the Flow without reregistering it with the Prefect API.')
#!/usr/bin/env python from logging import getLogger from datetime import timedelta from os import getenv from pathlib import Path from pendulum import today from prefect.engine.state import Failed from prefect.schedules import IntervalSchedule from prefect.storage import Docker from prefect.utilities.notifications import slack_notifier from prefect.utilities.storage import extract_flow_from_file logger = getLogger("dbt.deploy") with open('requirements.txt') as file: packages = list(line.strip() for line in file.readlines()) docker = Docker( registry_url="europe-docker.pkg.dev/projecthere/cd", # dockerfile='Dockerfile', # Uncomment to use own Dockerfile with e.g. dependencies installed image_name="analytics-dbt", image_tag="0.14.11", python_dependencies=packages ) slack = slack_notifier( only_states=[Failed], webhook_secret='SLACK_WEBHOOK_URL') every_hour = IntervalSchedule( start_date=today('utc'), interval=timedelta(hours=1)) flows = sorted(Path('flows').glob('*.py')) # Add flows for file in flows: flow = extract_flow_from_file(file_path=file) <http://logger.info|logger.info>('Extracted flow from file before build') docker.add_flow(flow) # Build storage with all flows docker = docker.build() # Update storage in flows and register for file in flows: flow = extract_flow_from_file(file_path=file) <http://logger.info|logger.info>('Extracted flow from file after build') flow.storage = docker flow.state_handlers.append(slack) flow.schedule = every_hour <http://logger.info|logger.info>('Registering...') flow.register( project_name='dbt', build=False, labels=['prod'], idempotency_key=flow.serialized_hash(), )