Hello, I have gotten this error when executing a flow in fargate

2019-12-13 1:40am	prefect.CloudFlowRunner	INFO	 Beginning Flow run for 'dbt_run' 
2019-12-13 1:40am	prefect.CloudFlowRunner	INFO	 Starting flow run. 
2019-12-13 1:40am	prefect.CloudFlowRunner	ERROR	 Unexpected error: TypeError("start() missing 1 required positional argument: 'self'")
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 400, in get_flow_run_state
    with executor.start():
  File "/usr/local/lib/python3.7/contextlib.py", line 239, in helper
    return _GeneratorContextManager(func, args, kwds)
  File "/usr/local/lib/python3.7/contextlib.py", line 82, in __init__
    self.gen = func(*args, **kwds)
TypeError: start() missing 1 required positional argument: 'self'

This was a fargate task started by the fargatetaskenvironment

At first glance it seems like something is happening where it’s calling

your_executor.start()

instead of

your_executor().start()

. As in the executor isn’t getting initialized

so in comparing environments remote vs fargate remote:

# Load serialized flow from file and run it with a DaskExecutor
            flow = storage.get_flow(flow_location)
            with set_temporary_config({"engine.executor.default_class": self.executor}):
                executor = get_default_executor_class()

            executor = executor(**self.executor_kwargs)
            runner_cls = get_default_flow_runner_class()
            runner_cls(flow=flow).run(executor=executor)

fargatetask:

# Load serialized flow from file and run it with the executor
            with open(
                prefect.context.get(
                    "flow_file_path", "/root/.prefect/flow_env.prefect"
                ),
                "rb",
            ) as f:
                flow = cloudpickle.load(f)

                runner_cls = get_default_flow_runner_class()
                executor_cls = get_default_executor_class()
                runner_cls(flow=flow).run(executor=executor_cls)

Have you ran into this before? I think that executor in the fargate task environment needs to be something like

executor=executor_cls()

yup

Yeah that needs to be updated

this is the first time we have run into it

so in our fork I added a new flag called

enable_task_revisions

for the fargate agent and fargate task environment that will leverage task definition revisions instead of creating a new task definition family every time

😮 amazing @Braun Reyes!

we want to battle test a bit...wasn't sure if we should wait to try and PR this after its in prod a bot and just go for it now

Whatever works for you! I’m excited to try it 🙂