Prefect Devs, I am trying to writing a custom process to push flow_runs to a group of worker via pub/sub style approach. I currently have a group of ProcessWorkers, but they trip over each other because they are polling from the same queue. I am looking for a way to push those flows to the ProcessWorkers, instead of them polling

Hello! Looking for some help on a problem I am running into. I have a parent flow that runs a bunch of subflows, it runs X of them at a time and does it in batches. However I want it to be executed in rolling concurrency, so theres always X running and when one finishes it launches a new one. I was able to get this to work with some test code (like a sleep x seconds then print) but I am running into an issue when trying to run product code. It seems like I need to do something like this where I launch the code as a new thread so that its async and not blocking the event loop - like this

async def my_subflow(id: str):
    def run_logic():
        product_logic(id)

    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, run_logic)

However - I am getting this error when trying to set up the logger:

Flow run encountered an exception. MissingContextError: There is no active flow or task run context

So some needed context is being lost when I create the new thread. Im wondering if there is a different approach for how I can execute the logic as a non blocking routine or maybe some way I can set the needed context? Thanks!

Hello fine people, I'm seeing an odd state transition fairly reguarly (at least once a day), we have a false negative "crashed" flow run, where the state transitions from:

Scheduled → Pending → Crashed → Running → Completed

For now, we've adjusted our alerting to only trigger if a flow "stays in {state} for 5 minutes", but it would be nice to be able to alert when a flow "enters {state}". lmk if there is a better forum for this bug report.

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?