In v2, is anyone packaging dependencies in storage along with the flow files? Seems like this could allow for dependencies per flow in a mono-repo without needing a separate docker image per flow. Kind of how aws lambda zip files work.

Hi, This is kind of difficult to explain. I have a flow that performs a job over an EC2 Instance, and the problem is that sometimes it fails and the most of times, it succeed. The job that runs, do the following operations: 1- Download a jar from somewhere. 2- Download some script from github. 3- Execute the script with some sort of arguments. And what it does, is download data from AWS S3, process the files downloaded, and then upload the results. The errors that I usually got are like •

Main class from jar could not be found

•

The output directory does not exists, so it could not be synchronize to AWS S3

Do you have any idea of why these types of issues happens time to time? BTW: I am using

boto3

SSM

agent to handle EC2 Instances creation, execution and termination

Hi - it seems when a flow is registered to Prefect Cloud (v1) with a schedule, then re-registered without a schedule (ie flow.schedule = None), the schedule is preserved server-side and will stick around until manually toggled off via the UI. Is this behaviour intended? If it is, is there a way to override this programatically by explicitly setting an empty schedule of some sort?

running into an issue where the flow runs show running status but the job has terminated on kubernetes . the agent pod reading from the queue has restarted so cant see the logs . please check the attached screenshots. cc @Taylor Curran

seeing this error first time.. anyone else who faced this an was able to fix it ?

07:01:33.898 | ERROR   | Task run 'run_executor-a1954751-160' - Crash detected! Execution was interrupted by an unexpected exception: AssertionError

quick question 🙏🏻 I built a work queue for Concurrency:

unlimited

It works perfectly fine. However, when I modify Concurrency into, for example, 3 It failed to run jobs 😞

one small question! is there a way to run deployment parallely?

Hi, I am trying to pass a list as a parameter via the UI. But it gets always as a string. Thank you

is it just me or parallel task runs (using the default concurrent task runs ) are failing left right and centre today

Hey there ! Starting to dip my toes into Prefect 2. Was wondering if there is an equivalent to Task classes we could do in Prefect 1 where the only requirement was to override the

run

method. It felt like a good way to encapsulate complex tasks and improve code readability. Creating abstract tasks was also something I did sometimes. Is this behavior gone or has it evolved ? I couldn't find much in the docs regarding this unfortunately.

Hi all -- has anyone seen an error like this before?

Encountered exception during execution:
Traceback (most recent call last):
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/engine.py", line 612, in orchestrate_flow_run
    waited_for_task_runs = await wait_for_task_runs_and_report_crashes(
  File "/software/projects/askaprt/tgalvin/setonix/miniconda3/envs/acesprefect2/lib/python3.9/site-packages/prefect/engine.py", line 1317, in wait_for_task_runs_and_report_crashes
    if not state.type == StateType.CRASHED:
AttributeError: 'coroutine' object has no attribute 'type'

I am running a known version of my workflow on a known dataset, which has worked perfectly fine dozens of times before. It seems to be saying the the

state

above is not an

orion

model -- rather a coroutine. All my tasks are using the normal

task

decorator around normal non-async python functions.

Hey guys. What’s the best practice for testing a flow that uses blocks? I am creating a block with credentials for a test minio storage and running

.save

on it, but when I try to retrieve it later by

load

I get error 404 from Prefect.

ValueError: Unable to find block document named test-minio-url for block type string

@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
    with prefect_test_harness():
        yield


@pytest.fixture(scope="session")
def minio_blocks(prefect_test_fixture):
    minio_creds_block = MinIOCredentials(
        minio_root_user=Config.MINIO_USER,
        minio_root_password=Config.MINIO_PASSWORD,
    )
    minio_creds_block.save("test-minio-creds")
    minio_url_block = String(Config.MINIO_URL)
    minio_url_block.save("test-minio-url")
    return minio_creds_block, minio_url_block


@pytest.fixture
def dummy_mission(minio_blocks):
    minio_creds_block, minio_url_block = minio_blocks
    minio_url = String.load(minio_url_block).value # <- ERROR HERE
    minio_url = minio_url.split("/")[-1:][0]
    minio_creds = MinIOCredentials.load(minio_creds_block)

Hello prefect community 🙂 I am curious if anyone here was able to run prefect behind nginx server.

When using s3 storage does a k8s job when running the flow just download the files from the s3 path indicated in the deployment rather than the whole bucket?

In prefect 2, how can I test a flow or task which is using the

get_run_logger()

which is set from the context. You can find sample test code on the thread The test keep failing with the erorr

prefect.exceptions.MissingContextError: There is no active flow or task run context.

Hello expert~! I'm using prefect 1.0 and looking method to visualize the task running progress at our own web page. I made a flow that post-processes data searched by a user on the web, and an event to download data from the web page triggers that flow. During processing on prefect, we want to show how far the task has progressed on our web page also, like the prefect UI shows the state of the task being executed. I found similar questions and answers at google, but it is impossible situation to divide the task into one per data. (link) Is there a nice way to check the progress of a long-run task on our web? Is there such a feature in prefect 2.0?

Hi. I need some quick help: We had some prefect 1 flows that were working fine on Monday and since Thursday they just freeze randomly. Liek one works, one just freezes. We are in the process of moving over to prefect 2, but still, we use 1.3.0 for what we have in prod now. I have been trying to debug it since it runs some ECS Tasks but I was not able to (larger machines, check AWS side of things etc). IT`s weird that the code (minus some print staements) worked perfectly fine on Monday. Any suggestions ?

Hi all! I installed prefect orion on a linux VM. I get an error when I open the web interface.Can't connect to Orion API at https://my_ip/api. Check that it's accessible from your machine. I used different IPs in the set PREFECT_API_URL command. I think I'm not the first with this issue. Help)

With prefect 2 how can I add dependence between mapped tasks? I want to skip any mapped downstream tasks if the corresponding mapped upstream task fails but without a direct link between the tasks. This is what I have so far:

# Prefect 2.6.9
# Python 3.8
from prefect import flow, task, get_run_logger

@task
def add_one(x):
    if x==1:
        raise Exception("Raised exception")
    return x+1

@task
def do_something(dummy):
    get_run_logger().info("Doing something")
    return

@flow
def mapped_flow_not_dependent():
    a = list([0,2,3])
    b = add_one.map(a, return_state=True)
    c = add_one.map(b, return_state=True)
    d = do_something.map(a, return_state=True, wait_for = [c])
    
    print(c)
    print(d)
    
    return "Flow completes"

if _name_ == "_main_":
    mapped_flow_not_dependent()

One state in c being failed means none of following do_something tasks run, whereas I would like all of the do_something tasks to run apart from ones where c is failed. I can get the desired behaviour by linking the tasks explicitly: changing the argument of do_something from a to c (and removing the wait_for kwarg).

Thanks @Kalise Richmond this was the case! In order to fix this I am trying to setup SLAs to kill flows stuck in running state via the automations section (using prefect 1) - doesn't seem to give me the option - I am on the standard plan which suggests this should be possible?

hey guys, I just read that in Prefect v1 there’s Prometheus integration for Tasks https://docs-v1.prefect.io/api/latest/tasks/prometheus.html#pushgaugetogateway, is the similar integration coming/available in Prefect v2?

I have a function that has a task decorator:

@task(cache_result_in_memory=False)

But I'm getting the error below that it's an unexpected keyword. Not sure what I'm doing wrong. It's running on prem using version 2.6.9.

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 883, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "/opt/prefect/processor.py", line 3, in <module>
    from data_import import data_import_process
  File "/opt/prefect/data_import.py", line 8, in <module>
    from data_cleaning import cleaning_process
  File "/opt/prefect/data_cleaning.py", line 42, in <module>
    @task(cache_result_in_memory=False)
TypeError: task() got an unexpected keyword argument 'cache_result_in_memory'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/prefect/engine.py", line 256, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/prefect/client.py", line 103, in with_injected_client
    return await fn(*args, **kwargs)
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/prefect/deployments.py", line 69, in load_flow_from_flow_run
    flow = await run_sync_in_worker_thread(import_object, str(import_path))
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 57, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/prefect/utilities/importtools.py", line 193, in import_object
    module = load_script_as_module(script_path)
  File "/opt/conda/envs/prefect/lib/python3.10/site-packages/prefect/utilities/importtools.py", line 156, in load_script_as_module
    raise ScriptError(user_exc=exc, path=path) from exc
prefect.exceptions.ScriptError: Script at 'processor.py' encountered an exception

Hey everyone, just want to confirm this: if I write (output1, output2) = task() in my flow, the task will be will run 2 times, and return the same output? (Prefect 1).

Hello All , I'm getting the following error while registering the flows ,the only change from the previous successful is we have installed the teradatasql connector on the docker image

Traceback (most recent call last):
    File "/usr/local/lib/python3.7/site-packages/prefect/cli/build_register.py", line 134, in load_flows_from_script
    namespace = runpy.run_path(abs_path, run_name="<flow>")
    File "/usr/local/lib/python3.7/runpy.py", line 263, in run_path
    pkg_name=pkg_name, script_name=fname)
    File "/usr/local/lib/python3.7/runpy.py", line 96, in _run_module_code
    mod_name, mod_spec, pkg_name, script_name)
    File "/usr/local/lib/python3.7/runpy.py", line 85, in _run_code
    exec(code, run_globals)
    File "/repo/src/flows/4i_ssp_bene_data_shrng/4i_ssp_bene_data_shrng.py", line 14, in <module>
    import snowflake.connector as sf
    File "/usr/local/lib/python3.7/site-packages/snowflake/connector/__init__.py", line 16, in <module>
    from .connection import SnowflakeConnection
    File "/usr/local/lib/python3.7/site-packages/snowflake/connector/connection.py", line 25, in <module>
    from . import errors, proxy
    File "/usr/local/lib/python3.7/site-packages/snowflake/connector/errors.py", line 18, in <module>
    from .telemetry_oob import TelemetryService
    File "/usr/local/lib/python3.7/site-packages/snowflake/connector/telemetry_oob.py", line 20, in <module>
    from .vendored import requests
    File "/usr/local/lib/python3.7/site-packages/snowflake/connector/vendored/requests/__init__.py", line 119, in <module>
    from ..urllib3.contrib import pyopenssl
    File "/usr/local/lib/python3.7/site-packages/snowflake/connector/vendored/urllib3/contrib/pyopenssl.py", line 50, in <module>
    import OpenSSL.SSL
    File "/usr/local/lib/python3.7/site-packages/OpenSSL/__init__.py", line 8, in <module>
    from OpenSSL import SSL, crypto
    File "/usr/local/lib/python3.7/site-packages/OpenSSL/SSL.py", line 19, in <module>
    from OpenSSL.crypto import (
    File "/usr/local/lib/python3.7/site-packages/OpenSSL/crypto.py", line 3232, in <module>
    name="load_pkcs7_data",
  TypeError: deprecated() got an unexpected keyword argument 'name'

Can prefect runs be triggered by some event rather just scheduled for a set time?

I want to set a memory limit on the docker container so that it will terminate instead of consuming all the memory on the server and crashing the server. Where can I set

--memory="[memory_limit]"

argument to enable this?

Hi folks! I'm converting my code base to use orion instead of v1, and I noticed that the ability to raise a SKIP signal (or something like that) is missing. In my original code, I have a sequence of tasks

A -> B -> C -> D

. Any of them can raise a SKIP signal, and if that happens, I want to skip all downstream tasks. It isn't a failure or a success, it's more of a "I'm not ready to run this yet, so don't run anything that depends on my output yet" Is that possible to do in Orion? I saw one earlier thread here about it, but the outcome was inconclusive... • one option is just to return a cancelled state, but that seems to suggest failure (which in my case would prompt a message to the admin, which I definitely don't want to happen for SKIPs. SKIPs happen very often in my particular case -- far more common than any other outcome) • another is to return a completed state, but then I need annoying

if

statements everywhere checking the outcome of previous tasks (skip vs. was actually run successfully). Actually, the whole reason I started using Prefect in the first place was for its ability to easily control flows where things get skipped 😉

Hi team, I have a flow containing multiple tasks, which are intensive in both computation and memory. Due to the essence of the business I want to handle, in one flow run, one task might be called several times for the same parameters or different parameters. To avoid the waste of resource and take the full usage of task result cache, I intend to limit concurrency of the task run with same parameters to 1. Other task runs with same parameters should wait in Pending or AwaitRetrying state. I have implemented this behavior in Prefect 1.x, by wrapping the exact task function with a Redis lock context manager. If a task run acquires the lock, then it could continue with the Running state. Otherwise, it raises a Retry signal, enters Retrying state, retries until it acquires the lock, and returns with cache. My query: is that possible to do this with Prefect 2.x? Can I manually turn a task run's state to Pending or Scheduled during flow runtime? (I read the code and I know Prefect uses a set of rules and core policies to manage the transition between states. I wonder if there is anyway to overwrite/inject rules from client side.) Also, I know we have concurrency limit control by tags in Prefect 2. But is there a feasible way to create/destroy the tags and concurrency limits attached to the tags based on "task name + parameters" during runtime? Much appreciated for any advice.

Hi Team, with prefect2.0 how to send task's error log to the slack on failure

Hi there ! Go some weird issue trying out flows locally. I'm running a local orion with

prefect orion start --port 5000

as well as the very simple flow copied below. Running the flow with

python log_flow.py

randomly ends up crashing with this error :

RuntimeError: The connection pool was closed while 2 HTTP requests/responses were still in-flight

. Is this an issue related to the server ? Am I forgetting to await something ? Anything I could do to fix it ?