Hi all, I am having a problem with creating an agent using prefect server as a backend. When I try to kubectl apply the agent manifest I get the following error:

prefect.exceptions.ClientError: Malformed response received from API.

Hello. Is there a resource on how to setup Prefect Server and Agents within AWS where Agents EC2 are in Private subnets and Server is setup on EC2 with Public subnet that covers security aspects in details?

Hi everybody, I'm working on a flow which should register another flow. And while flow.run() works perfectly fine for me, the running on the Cloud gives an error described in the picture. I've dived into this issue and realized that it does register another flow but outputs an error printing the success message))). So I tried changing print(msg) -> print(msg.encode('utf-8')) in the line 979 of the prefect/client/client.py and it helped me to fix this issue. Please consider adding this changing to the later versions of Prefect. Thank you in advance

Hi again, we have a requirement to handle two classes of errors: retryable errors and non-retryable errors. I think I saw some examples how this might be achieved somewhere but cant spot them for Prefect 2. In the non-retryable scenario we'd want the task state to go immediately to Failed (bypassing any retry settings configured on the task decorator)

I seemed like

prefect.tasks.kubernetes.job.RunNamespacedJob

doesn't report properly to prefect server. I didn't receive any log from spawned kubernetes job at all. Here are my screenshots

Hello everyone, I have a local agent which is not executing my flow run. The flow has been registered in the prefect server. The local hello world task can be executed locally but it would not update the state in the UI. If I run

prefect run -i "efc4568a-1fbd-4a14-8ef1-f137bdb45fcf" --watch

I receive the following warning:

WARNING | It has been 270 seconds and your flow run has not been submitted by an agent. Agent 122a102c-f083-48a8-b6f4-efd38e1b14f6 (agent) has matching labels and last queried a few seconds ago. It should deploy your flow run.

Hello, I’m new to Prefect and I need some help getting started. I’m currently struggling with two issues: 1. I’m setting up Prefect server on EKS using Helm. After

helm upgrade

ran successfully, I get commands such as

1 Run the following command to get the UI URL:

  UI_HOST=$( \
    kubectl get svc \
    --namespace prefect \
    --template "{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}" \
    prefect-server-ui \
  ) \
  && echo "UI available at: http://$UI_HOST:8080"...

when I run the above command, I get an error saying

error: error executing template "{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}": template: output:1:10: executing "output" at <index .status.loadBalancer.ingress 0>: error calling index: index of untyped nil

Then I have to run

kubectl port-forward <name of the service> <port>:<port>

to get ui and apollo servers running. Is there a way to run this without port forwarding? 2. I’m trying to run a hello world example flow on prefect server on my EKS cluster. I tried to trigger it from my command line by creating a project and running

prefect register --project test-project -p hello/

. The registering seem to run successfully but I don’t see the flow on the ui. I’m also not able to manually create flow project using ui. Am I interacting with two different prefect engine here? How do I register prefect flow so that it is running on my prefect server on EKS cluster?

Hi, I was wondering if anyone has tried using Redis as the storage backend for orion (prefect 2)? I see there is no built-in support to it. https://orion-docs.prefect.io/concepts/storage/ The closest we get is KV Server Storage. To use redis, does that mean I need to expose a HTTP API in between Orion and the redis server?

I have a flowrun with some curious logs. The logs claim that the heartbeat was not detected and the flow was cancelled, but according to the server it was still running and I cancelled it manually just now. The id is 99b7bd01-7b90-4634-9718-8e2eab1c00b0. Thank you.

Still learning Prefect concepts here and would love some advice. I'm trying to create a flow where I map a bunch of tasks. then when some of them fail, run a recovery task for each failing task, passing it the input of the original task to do an alternative action. At first it seemed like employing state handlers would help, then Kevin pointed me towards using the "any_failed" trigger instead, but it seems like that is built for a scenario where you have X tasks and you want to run 1 recovery task- but I have X tasks and want to run Y recovery tasks (somewhat less than X). It seems from the docs that prefect caches the inputs to a task, but I don't see how to access those inputs in a subsequent task. Is this possible? Thanks!

It seems Orion has an issue with the newest version of 'click' (8.1.0): "ImportError: cannot import name 'get_terminal_size' from 'click.termui' (/home/ubuntu/orionenv/lib/python3.8/site-packages/click/termui.py)". Downgrading 'click' to 8.0.3 (the version that I had elsewhere) made this error go away.

message has been deleted

Hello, can anyone help me? thanks 😄, I'm trying to create unit/integration tests to check my flows and tasks. I'm using Prefect Orion

prefect = "2.0a13"

, I did read this link, but I'm using async flows. More infos in the thread.

Hey gang. Today I released this new GitHub template repository. It includes all the bits necessary to deploy a simple application to Google Cloud and Prefect. Once the configuration is complete, you can ship changes with a simple GitHub release. https://github.com/biglocalnews/prefect-flow-template. There's a full tutorial in the wiki user manual.

All feedback welcome.

Hi folks. I'm trying to get my tasks to log to Cloudwatch through the ECSRun configuration. I'm testing out specifying a yaml config in

task_definition_path

, and running into an issue. My flows run on different ECR images, so I don't want to specify the image in the Task Definition file. However, when I leave it out, and specify it in the constructor:

run_config = ECS_RUN(image=ECR_IMAGE, task_definition_path='s3://<path>'

I get an error that container.image should not be empty. My task definition file that I'd like to use is:

networkMode: awsvpc
cpu: 1024
memory: 2048
containerDefinitions:
  - name: flow
    logConfiguration:
      logDriver: awslogs
      options:
        awslogs-group: test_prefect_log_group
        awslogs-region: us-east-1
        awslogs-stream-prefix: ecs-prefect
        awslogs-create-group: "True"

Is there any way to use a Task Definition file without specifying an image? Or is there another/better way to make sure my ECS tasks get logged? Thanks for any help.

Hi, trying to print out the

date

and

today

context in a task, but getting None.

date = prefect.context.get("date")
today = prefect.context.get("today")
def get_data():

    print(f"{date}")
    print(today)

hi all, i'm trying to write a flow where all the tasks should execute in a sequential order but some of them should optionally be skipped. basic example in normal python:

print("hello world 1")
if not skip_task2:
    print("hello world 2")
if not skip_task3:
    print("hello world 3")
print("hello world 4")

here's an attempt in Prefect:

@task()
def print_task(value: str) -> str:
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(value)
    return value


with Flow("case_skip_flow") as flow:
    skip_task2 = Parameter("skip_task2", True)
    skip_task3 = Parameter("skip_task3", True)

    task1 = print_task("hello world 1")
    with case(skip_task2, False):
        task2 = print_task("hello world 2", upstream_tasks=[task1])
    with case(skip_task3, False):
        task3 = print_task(
            "hello world 3", upstream_tasks=[task1, task2], task_args={"skip_on_upstream_skip": False}
        )
    task4 = print_task("hello world 4", upstream_tasks=[task3], task_args={"skip_on_upstream_skip": False})

i'm getting into trouble with the upstream skip status for

task3

. • if I set

skip_on_upstream_skip

to

True

, then when

task2

is skipped, so is

task3

regardless of the value for

skip_task3

• if I set

skip_on_upstream_skip

to

False

, then

task3

runs regardless of the value for

skip_task3

how can i implement a pattern like this?

Hi, I have a flow that contains a few docker related tasks (amongst many others) - PullImage, StartContainer, etc Works fine locally but proved to be tricky to run as a DockerStorage on a test vm. After some trial and error the solution I found how I can start a container from a container (in case someone else is stuck too):

flow.run_config = DockerRun(image="my_image", host_config={"binds":["/var/run/docker.sock:/var/run/docker.sock"]})

But I also need to pull an image when its not available and this step fails since I am not authenticated from inside my flow container. I can probably create a shell task that does docker login for me before image pull, but I am wondering if there are better options?

Hello, we have a prefect 2 server running a on GCP Cloud Run service - could you point me at some docs for how auth is setup, are there any options other than identity aware proxy for example (ideally for both front end and command line access)

Hi all, quick question: I can't find a mongodb crud task. Have any of you guys have something that works?

Hello all. Have a small question about cloud hooks. According to the documentation they work only at the flow level and not at the task level. One of the states I can select in the cloud hook is "TimedOut". However I can't find how to get my flow to enter that state. When I create my flow, I don't see any timeout condition for the entire flow run itself. Have I understood it correctly that cloud hooks only work at the flow run level? And if so how can I get my cloud hook to trigger when a flowrun enters the state "TimedOut"? Thanks in advance

Hi everyone. Am using a github action to register a flow. The flow uses S3 as storage and run_config is fargate config. The problem is, when registering it fails with an error related to boto3, which am not even using boto3, but before it was failing due to lack of boto3 library, so I added a command to install boto3 with pip. But now the error is "botocore.excwptions.ClientError: An error occured (AccessDenied) when calling the putObject operation: Acess Denied" Is this setup feasible? Why is it asking for boto3? Does prefect use boto3 library to use S3 as storage in the flow? Should I add auth to aws using github action or aws auth using boto3 in the flow python code itself?

Hi Everyone! I have a small question regarding dedicated execution of prefect tasks using a DaskExecutor. For example let's say I have some special tasks in a flow that need to be executed specifically on a dedicated machine. Is it possible to provide this information in Prefect?

Hey folks! We have a single agent running on EC2 and it has a few tags associated with it, some of which are associated with a flow concurrency limit. (e.g. etl_jobs => limit:5, ml_jobs => limit:2, and so on). Am I reading the Prefect 2.0 docs correctly? We would need to run multiple agents on our existing EC2 instance if we want to have different Flow concurrency limits for different groups of flows? We would have an “ETL Job” work queue with it’s won limit and a dedicate agent, another work queue for “ML Job” and a related agent.

In Orion, task run result storage is not implemented yet. Is there already a blueprint on what it will look like?

Does anyone know if it’s possible to find out the ECS task ID of a Prefect task from Prefect or the task itself? The context is I am using the ECSAgent to run a task which dumps its logs into a cloudwatch group which is called

prefect/task/{ECS-task-id}

. My aim is to display those logs in our webapp to show admin users the status of the flows they’ve started Ideally I’d be able to use the ID of a flow run and use that to query the ECS task it ran on and then use that to query the logs of the flow

I installed a dask cluster (https://docs.prefect.io/orchestration/recipes/k8s_dask.html#kubernetes-manifests) on our kubernetes cluster and I can run flows on it no problem using flow.run(). However, when I register the flow it tries to run the job as a kubernetes job. I have prefect server deployed and can see the runs, but they all fail with 'module C not found'. I am missing something, why do the flows not execute on the dask cluster?

Hi! I want to be able to set some Parameter when I register flows (or I guess replace the default value), is it possible?