Failed to load and execute Flow's environment: ValueError('Flow is not contained in this Storage')

[gryBox]

Hi - I am getting an error trying to deploy a flow with Non-Docker Storage for Containerized Environments using the FargateAgent and Cloud. The FargateAgent is hosted locally.

Error

Steps and code I used to invoke a Cloud quick run which produced the error.

1. Built and pushed the docker image to ECR

• Docker

FROM prefecthq/prefect:latest-python3.7

COPY ./ /MOC/
WORKDIR /MOC

ENV PYTHONPATH /MOC

RUN apt update && apt install build-essential -y build-essential libpq-dev postgresql-client postgresql-client-common && rm -rf /var/lib/apt/lists/*
RUN apt-get update && apt-get install curl -y

# Expose the PostgreSQL port
EXPOSE 5432
EXPOSE 80

RUN pip install pip --upgrade
RUN pip install "prefect[aws]"

RUN pip install .

2. Ran the deploy script that includes flow.register

• delpoy.py

from prefect.environments import LocalEnvironment
from prefect.environments.storage import S3

# Schedule
from prefect.schedules import clocks, filters, Schedule
from prefect.schedules import IntervalSchedule
from prefect.engine import signals
from datetime import timedelta
import pendulum

import etl_moc_flow

etl_moc_flow.schedule = Schedule(
    # fire every day
    clocks=[clocks.IntervalClock(
        start_date=pendulum.datetime(2020, 4, 22, 17, 30, tz="America/Toronto"),
        interval=timedelta(days=1)
        )],
    # but only on weekdays
    filters=[filters.is_weekday],

    # and not in January TODO: Add TSX Holidays
    not_filters=[filters.between_dates(1, 1, 1, 31)]
)

# Storage
# https://docs.prefect.io/orchestration/execution/storage_options.html#docker
etl_moc_flow.storage = S3(
    bucket="s3-prefect-flow-storage",
    secrets=["AWS_CREDENTIALS"],
    )

# docker_repo_path = st.result[docker_repo_path].result
image = f"{docker_repo_path}/{dckr_image_name}:latest"

etl_moc_flow.environment = LocalEnvironment(
    metadata={"image": image}
    )

#  Register 
pushlog = etl_moc_flow.register(
    project_name="market-on-close", 
    build=True,
    set_schedule_active=False
    )

print(pushlog)

3. Started the `FargateAgent successfully.

• fargate_a.py

from prefect.agent.fargate import FargateAgent

agent = FargateAgent(
    #
    enable_task_revisions=True,
    launch_type="FARGATE",
    taskRoleArn="<>",
    executionRoleArn="<>",
    family="get-tsx-moc",
    cluster="moc-tsx-clstr",
    networkConfiguration={
        "awsvpcConfiguration": {
            "assignPublicIp": "ENABLED",
            "subnets": ["<>"],
            "securityGroups": []
            }
    },
    cpu="512",
    memory="3072",
    containerDefinitions = [
        {
            "logConfiguration": {
                "logDriver": "awslogs",
                "options": {
                    "awslogs-group": "<>",
                    "awslogs-region": "us-east-2",
                    "awslogs-stream-prefix": "ecs"
                }
            },

            "image": "<>/etl-moc-img",
            "name": "etl-moc-img"
        }
    ],
    labels=["s3-prefect-flow-storage"]

)

agent.start()

Thanks in advance

[joshmeek]

@gryBox What version of Prefect are you on? On 0.12.0 the LocalEnvironment functions differently compared to what is currently on the master branch. If on 0.12.0 change that to RemoteEnvironment and then see if it works

[gryBox]

@joshmeek That did it!. The flow is running. But :) When I tried the same fargateAgent in a remote EC2 (and nohup). It did not look like the agent was responding to messages from Cloud.

And finally how do I know if I have a zombie task. One of them does not seem to finish.

[gryBox]

I have been running this flow for a couple of weeks manually successfully.

[joshmeek]

The tasks will be marked by Cloud's zombie killer (if enabled) after some period of time stuck in a Running state and no heartbeats have been received

[gryBox]

Thanks.