hi, when using prefect server i'm getting a restart loop with this error:

File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 276, in _catch_revision_errors
    raise util.CommandError(resolution) from re
alembic.util.exc.CommandError: Can't locate revision identified by '05ea6f882b1d'

Application startup failed. Exiting.
Server stopped!

full stack trace in thread happened to anybody else?

Hi, I am using Prefect Cloud, I created one deployment for one work-pool which has one work queue. Now in kubernetes, I created a Process Worker pod and I scaled it to 10 pods horizontally. I put 10 as limit of flows should run on each worker. Now if I create 100 flows, so technically, all 10 prefect worker (process type) pods should pick 10 flows and execute in paralelly. But that is not happening. There is any guide how to scale workers on single work-pool to execute flows parallelly?

Hi, I’m using prefect 2.13.0 on windows and noticed that if I try to retry a flow from the UI (either after a task has failed, I’ve paused the flow, or I’ve cancelled the flow) it will retry all tasks, even those that were in the “Completed” state. Any ideas how I can get the flow retry feature to only retry non-completed tasks?

Hi all! We've recently noticed our prefect postgres database has hit 20GB in size, which seems like a large size. Does anyone know if prefect is saving out task results / artifacts (note I have no explicit artifacts in any of my flows), and if there are ways to configure a retention policy (like to delete things after a week)?

Hey team, is there a prefect 2 equivalent for workers of the prefect 1 agent start

--no-pull

?

Hey everyone! I have a dataframe that is being passed between tasks in the flow and by using cache it is much faster and there is no need to persist it to disk. My question is, is it possible to clear that big dataframe from cache completely after the flow is done and not by setting cache_expiration? I read into refreshing the cache but that did not help since I do not repeat tasks. I also tried with

del dataframe
gc.collect()

But it does not get released from cache Any ideas?

Does anybody have a link to Server

Notifications

documentations. The link from the UI is sending me to 404 error. I would like to customize my slack message, however I can not understand how to extract flow data inside the configuration setup

Good morning and happy Monday. I have a quick questions in regards to Prefect 1.4. Is there a way to cancel alot of flows that are running without the usage of UI (or thru the UI). The number of scheduled flows are substantial so it's not possible to cancel them one by one thru the UI.

I am trying to run a Prefect flow in ECS with a prebuilt docker image. However, when the container is pulled the default command is being run; where would i configure the prefect flow entrypoint to execute? Also, what is happening under the hood when the flow run is executing? Particularly with regards to the parameters sent, what is organizing that and how is that information sent to the environment where the flow run is being executed?

Hi, everyone. I followed this tutorial () and was able to set up Prefect agent in ECS. However, I would like to use prefect server, instead of prefect cloud. Is there a tutorial or learning resource where I can learn how to setup prefect server in ECS? p.s: thanks @Taylor Curran and Mat for the tutorial. You did a great job!

hey folks... we're trying to suppress distributed.worker and distributed.core from our datadog logs, and have traced them to prefect dask. can anyone share how we might suppress any dask-related logs in prefect? we tried specifying the loggers to suppress by name and it hasn't worked...

Is there a channel for feature requests? Now that I have everything set up in ECS I have some suggestions, particularly around the build control plane 🙂

When I set up Prefect in ECS, I went ahead and created the prefect worker task definition, but I noticed that there is also a prefect task definition for the Prefect flow run execution. Was that created by Prefect?

I am seeing a weird issue with tasks and pydantic models. This is my example. The exclude_unset=True is not working once when the task is called. If I use user_json.fn() it works as excepted. Does anyone know why the @task decorator would be messing with the pydantic model and if a known workaround is available?

class User(BaseModel):
    """Pydantic model for the User."""

    email: Optional[str]
    firstName: Optional[str]
    lastName: Optional[str]


@task()
def user_json(user: User):
    """Should return a JSON representation of only the set fields in the User model."""
    print(user)
    # Outputs: email=None firstName='John' lastName=None
    print(user.json(exclude_unset=True))
    # Outputs: {"email": null, "firstName": "John", "lastName": null}
    return user.json(exclude_unset=True)


@flow()
def fl_create_user():
    """Creates a new user with only the first_name set and calls the user_json task."""
    user = User(firstName="John")
    print(user)
    # Outputs: email=None firstName='John' lastName=None
    print(user.json(exclude_unset=True))
    # Outputs: {"firstName": "John"}
    json_user = user_json(user)
    print(json_user)
    # Outputs: {"email": null, "firstName": "John", "lastName": null}

I'm running dependent subflows with

run_deployment

(run from a task). I notice that if one fails, then the downstream ones don't fail. How do I make subflow failures propagate to the parent flow? I'm using the

ConcurrentTaskRunner

. Some need to run concurrently with each other.

for an example: I want the ones without the

wait_for

argument to all run at the same time, while any Failure kills the whole thing.

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
import time


def fail_hook(flow, flow_run, state):
    print("Failed")
    raise Exception(state.message)


@task(log_prints=True, on_failure=[fail_hook])
def my_task():
    time.sleep(15)
    # some code here
    raise Exception("This task has failed.")  # this will set the task to Failed state


@task(log_prints=True, on_failure=[fail_hook])
def my_task2():
    time.sleep(25)
    return "sdfsdf"


@flow(
    task_runner=ConcurrentTaskRunner(),
)
def my_flow():
    result1 = my_task.submit(return_state=True)
    result2 = my_task2.submit(return_state=True)
    result3 = my_task.submit(return_state=True)
    result4 = my_task2.submit(return_state=True, wait_for=[result1])

hi all, i have a question about whether prefect is the tool I'm looking for. I'll post the description inside the thread - it's on the longer side, looking forward to hearing your thoughts marvin

We are running prefect for 90 days in production. Today we noticed that the Prefect Server is querying Postgresql at a rate of 14MB/s. Grew from 3MB/s linearly over a course of a month. Is this something you are aware of? We nuked the db now and checking if this is related to the number of flows we ran so far.

Hi All, i am a newbie i have setup the dev environment when i am trying to run the command "prefect dev start" i am able to go to the local host link.But i am not table to see any UI it is showing blank page for me. Please help me on this. Thanks,

I am using this in debug mode

Are there examples of using 1 flow to run 2 subflows in Prefect 2.0? The documentation is not very clear on the matter. How do I test it locally? On my firm's local server?

Hi all, has anyone experienced the following issue with the prefect worker utilizing Kubernetes failing to submit flows to the infrastructure after running fine for several hours? A restart is required in order for the worker to function again. Here's the error trace

Failed to submit flow run '8093ef5c-64f8-42eb-9f4b-045ed7e7d6ad' to infrastructure.
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/urllib3/connection.py", line 174, in _new_conn
    conn = connection.create_connection(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/util/connection.py", line 95, in create_connection
    raise err
  File "/usr/local/lib/python3.11/site-packages/urllib3/util/connection.py", line 85, in create_connection
    sock.connect(sa)
OSError: [Errno 99] Cannot assign requested address

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 714, in urlopen
    httplib_response = self._make_request(
                       ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 403, in _make_request
    self._validate_conn(conn)
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 1053, in _validate_conn
    conn.connect()
  File "/usr/local/lib/python3.11/site-packages/urllib3/connection.py", line 363, in connect
    self.sock = conn = self._new_conn()
                       ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connection.py", line 186, in _new_conn
    raise NewConnectionError(
urllib3.exceptions.NewConnectionError: <urllib3.connection.HTTPSConnection object at 0x7f60db485950>: Failed to establish a new connection: [Errno 99] Cannot assign requested address

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 834, in _submit_run_and_capture_errors
    result = await self.run(
             ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 506, in run
    job = await run_sync_in_worker_thread(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 91, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/to_thread.py", line 33, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 877, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 807, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 628, in _create_job
    job = batch_client.create_namespaced_job(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api/batch_v1_api.py", line 210, in create_namespaced_job
    return self.create_namespaced_job_with_http_info(namespace, body, **kwargs)  # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api/batch_v1_api.py", line 309, in create_namespaced_job_with_http_info
    return self.api_client.call_api(
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 348, in call_api
    return self.__call_api(resource_path, method,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 180, in __call_api
    response_data = self.request(
                    ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 391, in request
    return <http://self.rest_client.POST|self.rest_client.POST>(url,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/rest.py", line 276, in POST
    return self.request("POST", url,
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/rest.py", line 169, in request
    r = self.pool_manager.request(
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/request.py", line 78, in request
    return self.request_encode_body(
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/request.py", line 170, in request_encode_body
    return self.urlopen(method, url, **extra_kw)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/poolmanager.py", line 376, in urlopen
    response = conn.urlopen(method, u.request_uri, **kw)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 826, in urlopen
    return self.urlopen(
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 826, in urlopen
    return self.urlopen(
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 826, in urlopen
    return self.urlopen(
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/connectionpool.py", line 798, in urlopen
    retries = retries.increment(
              ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/urllib3/util/retry.py", line 592, in increment
    raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='172.30.0.1', port=443): Max retries exceeded with url: /apis/batch/v1/namespaces/prefect-2x/jobs (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7f60db485950>: Failed to establish a new connection: [Errno 99] Cannot assign requested address'))

Question regarding a deployment's default parameters. If I run a deployment with

run_deployment

, are only the parameters I submit overridden, or are all parameters overridden and any not explicitly passed fall back on the flow function's defaults?

Hi all. I have been dealing with an issue related to the prefect-ray integration that i am not able to solve. Hope someone here can help me 🙂 I have a prefect agent and Ray cluster running inside my kubernetes cluster. Both seems to work as intended as a standalone. I.e i can submit jobs to the Ray cluster and i can also run simple prefect flows. However, when i try to use the prefect-ray integration and specify resources for a task i.e.

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def process(x):
    return x + 1


@flow(task_runner=RayTaskRunner("ray://<my-ray-service>:10001"))
def my_flow():
    # equivalent to setting @ray.remote(num_cpus=4)
    with remote_options(num_cpus=4):
        process.submit(42)

It fails with a runtime error:

RuntimeError: There is no current event loop in thread 'ray_client_server_1'.

If i do not specify resources for my task, everything works as intended. I have submitted an issue, but it has not been picked up yet, so i thought maybe i could get some help here 🙂 https://github.com/PrefectHQ/prefect/issues/10542

Has anyone run into this error? Only happens when we use the Dask executor, always at the same spot (right before the first mapped task). Also only happens in kubernetes, not when running locally.

Crash detected! Execution was interrupted by an unexpected exception: PrefectHTTPStatusError: Client error '408 Request Timeout' for url '<https://api.prefect.cloud/api/accounts/><xxx>/workspaces/<xxx>/task_runs/'

Hi all! I noticed that theres a preference for workers over agents now, but workers dont support storage blocks. Does this mean storage blocks are no longer recommended? If I've stored my flows in an s3 bucket, does this mean there's no way to use them with a worker? Should I be doing something completely different? We've got two use cases for flows: one set that runs tasks in a dask task runner that simple data ingestion and transformation, and another use case where we use

prefect-aws

to launch ECS tasks for longer running compute. Im unsure how a "worker" would change or improve these two use cases

Hrm, logging from within a Flow Hook isn't working? (Running on ECS)

def make_fail(flow, flow_run, state):
    print("Failed")
    flow_run_ctx = FlowRunContext.get()
    print(flow_run_ctx)
    print("Flow")
    print(flow)
    print("Flow run")

@flow(
    log_prints=True,
    on_completion=[make_fail],
    on_failure=[make_fail],
    task_runner=ConcurrentTaskRunner(),
    result_serializer=JSONSerializer(),
    result_storage=S3(bucket_path=bucket_path),
    persist_result=True,
)
def my_flow():
    result1 = my_task.submit()
    result2 = my_task2.submit()
    result3 = my_task.submit()
    result4 = my_task2.submit(wait_for=[result1])

None of the

print

statements show up

are there any means to query cached task runs the same way you might in a database?

Hi everyone ! I have recently started using prefect. Currently, I want to do some scraping with prefect as an experiment. My prefect is set up as prefect-server in kubernetes. How can I do the following?　 - Install the package with the configuration described in POETRY. (Or I would like to know the best practice for installing packages.) - Upload chromedriver. (Is it common to upload local files directly or mount persistent volume?)

are there known issues with the

/task-runs/task-run/:uuid

route? • the task inputs tab is showing

{"id": []}

when it should be (for example)

{"id": "123"}

• the cache key in the sidebar is not populated, but I provided a

cache_key_fn

and the caching feature itself seems to be working (re-running task with same ID uses cached result)