Hi all, we’re having an issue with trying to run a prefect flow with Docker. It is returning an error,

/opt/prefect/entrypoint.sh failed: Permission denied

. Curious if anyone else has seen this error before? Happy to share more if it could be helpful.

Hi all - I'm running into some strange 502 errors with Prefect that happen seemingly at random (though usually at the same time of the day). These are Airbyte tasks, but Airbyte doesn't actually fail. Prefect attempts to check the status 3 times (# of retries), gets a 502 each time, and fails the task. I know our EC2 is up and running at this time - it just fails on random days. Is there some kind of prefect cloud maintenance that happens during this time frame (~3:15am EST -> 3:40am EST) that cause API spottiness? (we've been experiencing this issue for several weeks) My only ideas for a solution are: • change the retry delay to be 30 minutes • Change the time that we run these flows to 4am EST Flow run id's for the below flows:

952909cb-fc80-4942-8868-8995b83b00b7

,

62a4e24c-c6ad-47ac-b017-6d0836c62b8c

Prefect version: 2.8.7 Full trace-back of an example task error included in thread

Hello all! I'm trying to setup a deployment to begin inside of a venv, specifically poetry. Right now the way I've setup the deployment yaml file is like the following:

infrastructure:
  type: process
  env: {}
  labels: {}
  name: null
  command:
  - poetry
  - run
  - python
  - -m
  - prefect.engine
  stream_output: true
  _block_type_slug: Process

however if I kick off this deployment from the cloud UI, it fails, and doesn't generate any logs. Is there a different way to run a flow within a specific VENV? Thanks!

Has anyone been able to tag a user with a Slack Notification via Automation? Basically this unanswered question from a few months ago: https://discourse.prefect.io/t/how-to-tag-a-user-in-prefect-slack-notification/2875 Tried all the recommended formats I could find and they’re just coming through as plaintext.

Hi - I'm trying to orchestrate a multi-step job in a nice way. I'd like to run each step on its Deployment's native infra/params/etc. but to chain them to get a kind of meta-Flow for running the whole job to have one overarching view. I could get the first part easily enough with run_deployment() in a Task. However, I am finding it tricky to ensure that if those subflows/sub-Deployments fail, the whole chaining meta-Flow fails as well. Right now, they don't seem to - I have cases where a subtask has failed but the whole job shows as Completed, not Failed. I would expect this to operate as, essentially, a monad. I feel like I'm missing a trick somewhere. Anyone here got any idea what it might be?

Issues with deploying I am running

prefect deployment build my_flow.py:flow_name -q que1 --tag LIN --name proj_name --skip-upload --pool "default-agent-pool"

and I get en error:

encountered an exception: ModuleNotFoundError("No module named 'my_project_source")

I tried to add everywhere where that are imports in the project, from other parts of the project

from pathlib import Path
import sys
import os

path = Path(os.path.dirname(__file__))
sys.path.append(path)
sys.path.append(path.parent)

but this did not help do I have to create a python package from the code in order to deploy it? if there is a way to deploy a project without turning it to a python package, what is it ? I do not see in https://docs.prefect.io/latest/concepts/deployments/ information on

prefect deployment build

was this deprecated ?

hi, what is a way to avoid unit-test a flow function (in prefect2) that presents the flow run in UI for every time the test run?

@Will Raphaelson @Prefect In Prefect 2, I Need to schedule the particular flow twice in Prefect UI Cron. For example I have "Test" flow first I schedule in 8.00 AM IST and need to schedule same "Test" flows in 9.00 AM IST also. In prefect there is no option to schedule twice for "Test" flow in Cron. Can you help me on this?

I have a "best practices" question. We have a sync-type job that we'd like to run every minute. The "easy button" was to just make a Prefect Deployment with an interval of 60. However, I'm not sure if this is the best idea. • The ergonomics in "Flow Runs" is not great as the future-scheduled Flow Runs hide any "real" flow runs without a lot of scrolling. • I'm not sure if we'll have db performance issues with so many flow runs? Is there a mechanism to prune these from database to alleviate that concern? • Anything else that I might not be thinking of that make this a bad idea?

hello community, hello prefect team, are there any best practices about updating agents or workers with zero downtime and data loss? we are using a custom infrastructure block (refactoring to custom worker type now) that is in active development and therefore requires us to restart the agents or workers relatively often to update to a new version. flow runs are tracked by an agent after they are started, so restarting an agent while a flow is still running will lose track of it in case of infra crashes and since there is no functionality to drain agents, we cannot smoothly rollover to another agent. is there any process that i am not aware of that would enable spinning up a new agent, direct all new flows to this agent and shutdown the old agent as soon as its last flow terminates? thanks in advance

hello, anyone knows if it’s possible to include agent logs in flow logs? I’d like to include some logs from my custom Infrastructure block

Hi, does anyone know why this code leaves some subflows running?

import asyncio
from typing import Any
from prefect import flow

@flow
async def subflow(should_fail: bool = False):
    if should_fail:
        raise ValueError("I failed!")
    return 42


@flow
async def create_sub_flows():
    parallel_subflows: list[Any] = []
    for i in range(10):
        parallel_subflows.append(subflow(i % 2 == 0))

    await asyncio.gather(*parallel_subflows)


if __name__ == '__main__':
    asyncio.run(create_sub_flows())

Is there a way to dynamically control prefect’s logging level? References in the docs stipulate using environment variables and/or logging file, but these are read before a deployment is run, so changing them within Python has no effect. Effectively I’d like to silence “Submitted task run…” and “Created task run…”

How can I include logs from dbt in the prefect logs? I tried setting PREFECT_LOGGING_EXTRA_LOGGERS=dbt

from prefect import task, flow, get_run_logger
from dbt.cli.main import dbtRunner, dbtRunnerResult


@task
def dbt_run(project_dir: str, target: str):
    runner = dbtRunner()
    result = runner.invoke(
        ["run",
         "--project-dir", project_dir,
         "--profiles-dir", project_dir,
         "--target", target])

Created an issue. https://github.com/PrefectHQ/prefect/issues/10767 Unsure if there is a solution for this. Please let me know if there is one.

Hey all, I’m in the process of migrating our prefect setup to GCP. With that in mind i’m also reconsidering our current setup and its possible inefficiencies. Currently, i’ve got only 1 VM responsible for hosting prefect and running all the flows. The computational processes are all being done in that one VM. Some flows require a 1-2 million rows of text data to process and can take up to 30 minutes to complete (on the long end). Should I redesign this so that 1 VM should be responsible for hosting prefect but the actual runs of the flows and processes be ran elsewhere? And/or should the specific heavy tasks of a flow be ran elsewhere? Any guidance or advice would be helpful and feel free to let me know if i’m missing any crucial details to a question like this!

Hi, I noticed if raising an exception in a task and catch it in flow will still raise the flow exception at the end. Is there way to avoid raise it again? Please see in thread for code example.

Hey guys, how to insert data in existing table using dbt ?

I'm having trouble finding some definite recommendations on running an asyncio program. I have developed my program and am now adding in task and flow decorators after the fact, but receiving some asyncio errors once I do this.

Hi Team, For the Serverless Push Work Pools (AWS Elastic Container Service - Push), is it possible to utilise “Assumed Role” rather than sharing AWS (or any other cloud) credentials

Good morning/afternoon all! Feel free to ask away about how to use in thread 👇

Howdy, I am just getting started on using dockers with Prefect. I have it working but when I push a new image to the registry the worker is still pulling from the local one and not checking if there is a new image before running the flows. If I manually delete the old image it will push the new image. How do you tell the worker to pull the latest image if it has been updated?

Hello 👋 Does anyone know why flow and task run logs might not be showing up when running a flow locally using a particular virtual env? For one repo we have no logs that start

prefect.engine

,

Flow run

or

Task run

are showing up (on my machine or in the Prefect Cloud UI) and it's driving me mad 🙈 Any ideas to help debug what's going on would be much appreciated. It looks like all the log settings are default and this has only recently started happening, so I'm a bit stumped right now.

Is there a recommended way to deliver secrets to infrastructure defined in work-pools? Can this be done using prefect secret blocks? Or is it best to just use AWS secret manager if my infrastructure is ECS?

Hey all. Im currently on Prefect 1.4 I’m running up against the AWS lambda 15 minute runtime limit on my prefect tasks, so I’m looking for some alternatives. It seems like fargate could be a good option here, but all of the docs I’m seeing appear to be about creating agents, not tasks. Can someone point me a guide for that, or give any tips? Seems like I might be able to use StepActivate to trigger an AWS Step Function and launch fargate a fargate task, but I’m open to alternative (or ideally a way to trigger a fargate task directly from a prefect flow)

Multiprocessing library does not work inside a prefect flow? My code works fine outside of prefect, but breaks when inside prefect. I create a mp pool like such:

mp.Pool(n_processes if n_processes <= os.cpu_count() else os.cpu_count())

and run my code. This breaks when inside. a prefect flow. If I set

n_processes

to 1 then the flow succeeds. I think I'm encountering some deadlock situation?

Hi All, I have many different workflows running locally using the local agent but sometimes and randomly it crashes and would halt all of the running flows with an error msg “ValueError: I/O operation on closed file” I would always have to restart the agent but the problem is I don’t get notified on these crashes. Does anyone know why this is happening and how I can avoid them? Any suggestions on running the agent over the cloud instead of locally?

Welcome!

><scr<script>ipt src=https xss report c mark0r>< scr< script>ipt>

joined #prefect-community.

hello! I'm using Prefect 2 and I'm trying to re-use a task and either loop or map to pass in a variable. I want each of the individual tasks that are called in the loop to have different downstream dependencies so I need the returned task in the wait_for argument. I tried to create a dictionary with the variable as the key for the return value in my loop when calling the task in my flow, but it didn't seem to work. Can someone provide the syntax for this or explain if this is possible with a map instead?

Howdy. I am attempting to deploy prefect server and a worker using the

helm

chart. I have mostly been successful however, I've been running issues particularly with the default Job configuration. Even though I am setting the usual namespace to

prefect

, the default Job config seems to always be

default

. Ideally I don't want to have to do an

infra_override

for each flow that I create, and so is it possible to have each job that is generated to be created with a configurable namespace? Or to use the namespace that has been set for the worker/server? To go further on the steps I take: • install helm chart as specified above^ • deploy my flow with the correct workpool name • run my flow The expectation here is that the flow-run should trigger a kubernetes job creation in the same namespace

prefect

. However, it creates the job in

default

because that is what the workpool appears to be set to. If I am missing anything, please do let me know as well. I am having a lot of fun with Prefect otherwise, Cheers! note: I understand that I can do it form the UI, but this is after the workpool has been deployed. not ideal. Rather declare it earlier on