Hi Prefect team. Apart from the issues mentioned before by Krzysztof and Thomas, we also see that some flows are not actually executed. It seems so from the parent flow, with a duration of 30 seconds, and the next dependent flows are triggered, but we cannot find the flow run when accessing the child flow directly. Besides, the UI shows both a “success” and a “failed” status

same here...

We are seeing even weirder behavior now, one flow was running one of its tasks twice (in parallel). Fortunately we have duplicate email detection on our notification gateway, but this is really concerning.

I'm trying to change PREFECT_API_URL by running

prefect config set PREFECT_API_URL=<https://example.com/api>

. However when I try to run the server it defaults back to http://127.0.0.1:4200/api

Hey I created a secret block on prefect cloud ui,

secret-block = Secret.load(“api-key”)

I want to use this block in the body of my flow how can I go about this? Should I just pass

secret-block

? Do I need to use the get method anywhere?

Hello, I see there is way to limit concurrency for a task using tags. But what about flows? How to make sure the same flow (or the same deployment) does not run concurrently?

Is there any nice way to modify the arguments given in the

@task

decorator after that task has been defined? Say I use a task from a collection that and want to use a custom

cache_key_fn

. It looks like i could do it by

from copy import copy
from some_collection import some_task
some_task_cache = copy(some_task)
some_task_cache.cache_key_fn = my_cache_fn

but that seems a bit hacky. The actual problem is the other way around - I want to provide a collection for users, but they would want to modify the task decorator arguments.

@Anna Geller I’m still working on implementing the aws-lambda-serverless-flow template example. On a closer look at the repo, I can see that you didn’t build deployment with prefect(no deployment scripts). Was this intentional or an omission? My attempt to build and apply deployment led to this error:`File "/home/ec2-user/.servy/lib/python3.9/site-packages/botocore/handlers.py", line 278, in validate_bucket_name` raise ParamValidationError(report=error_msg) botocore.exceptions.ParamValidationError: Parameter validation failed: Invalid bucket name “s3“ Bucket name must match the regex “^[a-zA-Z0-9.\-_]{1,255}$” or be an ARN matching the regex “^arn:(aws).:(s3|s3-object-lambda):[a-z\-0-9][0 9]{12}accesspoint[/:][a-zA-Z0-9\-.]{1,63}$|^arn:(aws).*s3 outposts[a-z\-0-9]+[0 9]{12}outpost[/][a zA Z0 9\ ]{1,63}[/]accesspoint[/:][a-zA-Z0-9\-]{1,63}$”

Hello, recently (within the past week) I have been unable to delete my flow runs (old and new) in Prefect 2 Cloud. I receive a

Failed to delete flow run

message in the UI. Is this a known issue? Or any thought as to why this occurs?

Hi All... I am trying to pull detailed prefect logs using the graphql, however i could not find the table/column name in graphql which holds the detailed logs. Any help would be really appreciated

Hi All, I am working on building profiler(to measure the CPU, Memory and Network IO utilizations) to attach it to any prefect2 task. Any ideas about where I can start from, does ORION collect those kind of data, so we can grab it, or need to create something from scratch?

Hello all, love the project, one question: I find myself really wanting a feature flag to turn task level result caching on and off. My use case seems common enough: if I update my ETL code, I want to invalidate caching so that new results on existing input data are produced. Keying the cache on input argument hash doesn’t solve this use case; the input data is identical, it is my code that has changed. Time based cache expiration doesn’t fit this use case; I don’t know when my code will be updated, it might be tomorrow, it might be in a year. It seems like prefect 1.0 had some options in the API to serve this use case. What am I misunderstanding about prefect 2.0, which as far as I understand has no clear way to address this?

Hi all, In Prefect 2, is it possible to schedule a deployment flow with custom input parameters for each schedule in UI/CLI? Or do I need to create a deployment for each set of parameters? Thanks!

is there a way to associate arbitrary data to orion objects? i.e. git repo, sha, contact team, etc may all be useful from a discoverability perspective on all flows/deployments

Hi, I'm using Prefect v1 and I'm having trouble with configuring timeouts for

ShellTasks

. Setting timeout like it says in the documentation doesn't have any effect, the task doesn't get terminated after 10 seconds in this case. Is that a bug in Prefect or is it an issue on my part and I'm somehow misconfiguring

ShellTask

? I really appreciate any help on this and thank you in advance!

Hi Prefect, We’re still in our journey to evaluate prefect, one question that comes to mind How do we run a flow that takes input parameters ? I mean without deploying to a agent, simply by running it with the command line

Hi all - I am working through the Getting Started tutorial for Deployments using Prefect 2.4.5 via a Miniconda environment on Windows:

name: prefect-service
channels:
  - defaults
dependencies:
  - python=3.9.13
  - pip=22.1.2
  - pip:
    - prefect==2.4.5

I am hitting an error after running

prefect deployment apply log_flow-deployment.yaml

:

Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x00000226A897BEE0>
... redacted for readability ...
RuntimeError: Event loop is closed

Any idea what is happening here? Please let me know if I can provide any additional info.

Hi everyone. We're trying to update our prefect v1 from 1.2.2 to 1.4.0. Since we operate the infrastructure on our own, we use the official helm chart to deploy everything. That part works just fine. Our services make sure that they work with the correct version of prefect when they are about to start after an upgrade in the k8s cluster. This check makes sure that Prefect Server version matches the used prefect library. We use following graphl query to fetch the API version:

query { api { core_version } }

We ran into a problem with version 1.4.0 when prefect API reports version "1.3.0+13.g4142953a5". Is that intended?

Hi everyone, I am trying to figure out how to set up a job to only run the first 6 days of the month. I was looking at the documentation below, but still having some challenges, wondering if anyone could give some insight on how to do this. https://docs-v1.prefect.io/core/concepts/schedules.html#clocks

Hello, I'm running dbt job with the agent on Kubernetes (Prefect 2.4.5 Python 3.10.5 prefect_dbt 0.2.2) and also using Prefect Cloud. The dbt job finishes, but the flow

trigger_dbt_cloud_job_run_and_wait_for_completion

fails with

AttributeError: 'NoneType' object has no attribute 'call'

. Here is the log:

21:33:15.658 | INFO    | Flow run 'charcoal-harrier' - Created subflow run 'hot-civet' for flow 'emails-gold'
21:33:15.958 | INFO    | Flow run 'hot-civet' - Created subflow run 'cocky-pony' for flow 'Trigger dbt Cloud job run and wait for completion'
21:33:16.069 | ERROR   | Flow run 'cocky-pony' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 589, in orchestrate_flow_run
    result = await flow_call()
  File "/usr/local/lib/python3.10/site-packages/prefect_dbt/cloud/jobs.py", line 287, in trigger_dbt_cloud_job_run_and_wait_for_completion
    run_id_future = get_run_id.submit(triggered_run_data_future)
  File "/usr/local/lib/python3.10/site-packages/prefect/tasks.py", line 492, in submit
    return enter_task_run_engine(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 741, in enter_task_run_engine
    return flow_run_context.sync_portal.call(begin_run)
AttributeError: 'NoneType' object has no attribute 'call'
21:33:16.133 | INFO    | Flow run 'cocky-pony' - Created task run 'Trigger dbt Cloud job run-e4229069-0' for task 'Trigger dbt Cloud job run'
21:33:16.134 | INFO    | Flow run 'cocky-pony' - Submitted task run 'Trigger dbt Cloud job run-e4229069-0' for execution.
21:33:16.264 | INFO    | Task run 'Trigger dbt Cloud job run-e4229069-0' - Triggering run for job with ID 135917
21:33:16.558 | INFO    | Task run 'Trigger dbt Cloud job run-e4229069-0' - Run successfully triggered for job with ID 135917. You can view the status of this run at <https://cloud.getdbt.com/#/accounts/xxxxx/projects/156718/runs/86440872/>
21:33:16.683 | INFO    | Task run 'Trigger dbt Cloud job run-e4229069-0' - Finished in state Completed()
21:33:16.812 | ERROR   | Flow run 'cocky-pony' - Finished in state Failed('Flow run encountered an exception.')
21:33:16.812 | ERROR   | Flow run 'hot-civet' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 596, in orchestrate_flow_run
    result = await run_sync(flow_call)
  File "/usr/local/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 "/usr/local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/opt/prefect/flows/emails_deploy.py", line 35, in emails_gold
    run = trigger_dbt_cloud_job_run_and_wait_for_completion(
  File "/usr/local/lib/python3.10/site-packages/prefect/flows.py", line 390, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 163, in enter_flow_run_engine_from_flow_call
    return run_async_from_worker_thread(begin_run)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 137, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/usr/local/lib/python3.10/site-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 458, in result
    return self.__get_result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.10/site-packages/prefect/client/orion.py", line 82, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 522, in create_and_begin_subflow_run
    return terminal_state.result()
  File "/usr/local/lib/python3.10/site-packages/prefect/orion/schemas/states.py", line 145, in result
    raise data
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 589, in orchestrate_flow_run
    result = await flow_call()
  File "/usr/local/lib/python3.10/site-packages/prefect_dbt/cloud/jobs.py", line 287, in trigger_dbt_cloud_job_run_and_wait_for_completion
    run_id_future = get_run_id.submit(triggered_run_data_future)
  File "/usr/local/lib/python3.10/site-packages/prefect/tasks.py", line 492, in submit
    return enter_task_run_engine(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 741, in enter_task_run_engine
    return flow_run_context.sync_portal.call(begin_run)
AttributeError: 'NoneType' object has no attribute 'call'
21:33:16.892 | ERROR   | Flow run 'hot-civet' - Finished in state Failed('Flow run encountered an exception.')
...

Noob question Sorry to post this here, but since its part of the Prefect docs it seems appropriate. I'm doing this tutorial from the blog Workflow without DAGs, a little confused by an error from the first code snippet

p102-13.py

If I leave

.result()

in the code, it fails with error:

letter = get_letter().result()
AttributeError: 'str' object has no attribute 'result'

But remove the

.result()

bit and I get success:

poetry run python src/quickstart/p102-13.py
16:48:31.385 | INFO    | prefect.engine - Created flow run 'jasper-uakari' for flow 'Result Flow'
16:48:31.454 | INFO    | Flow run 'jasper-uakari' - Created task run 'Get a letter-c414aff6-0' for task 'Get a letter'
16:48:31.454 | INFO    | Flow run 'jasper-uakari' - Executing 'Get a letter-c414aff6-0' immediately...
16:48:31.468 | INFO    | Task run 'Get a letter-c414aff6-0' - A
16:48:31.478 | INFO    | Task run 'Get a letter-c414aff6-0' - Finished in state Completed()
16:48:31.478 | INFO    | Flow run 'jasper-uakari' - You got: A
16:48:31.488 | INFO    | Flow run 'jasper-uakari' - Finished in state Completed('All states completed.')

The poetry environment is:

[tool.poetry.dependencies]
 python = "^3.10"
 prefect = "^2.4.5"
 greenlet = "^1.1.3"

Hi, i'm a prefect 1.0 user.🤗 We made a program called 'Uploader'(hereinafter referred to as 'uploader')that can be used outside the server, and the function of that program is to record the meta information of the data in the database installed on the server. What I want to do that the prefect flow is executed as soon as the program has finished writing to the database (if the program has performed all the tasks it needs to do). Until now, after the uploader program execution was finished, a person directly executed the prefect flow. There is a way to set this flow as a schedule and run it according to the set schedule regardless of whether the uploader is terminated, but I want to avoid the execution by scheduled because the uploader is not periodically used. What are the appropriate conditions to run the prefect flow automatically after ended the uploader's process? (It's as long as what I wrote above, but...) To summarize, • There is a program(uploader) that works outside the prefect server. • Uploader functions to send data to server by the user and records the program's log and meta information about the data in a specific table of database. • When the execution of uploader is completed, prefect executes flow to preprocess data newly added by the user. • (As-is) When the uploader's task is completed, someone starting the flow of the prefect directly. • I would like to avoid scheduling the flow because the uploader is not used periodically, if we can do it. Isn't there any way out to executing the flow automatically after than end of uploader process?

Hi, I have a question about concurrency. I have a queue with concurrency = 1. Yesterday I manually started a job run via UI and also I have scheduled the same flow to run in 10 hours or so. The manually triggered job didn't finished in 10 hours, so another one shouldn't have started (because concurrency = 1), but it did start. Am I doing something wrong?

I was browsing the Prefect codebase trying to understand how notifications are implemented, when I came across how services are implemented. Having injectable/modular services would be awesome, so I create a feature request: https://github.com/PrefectHQ/prefect/issues/7064

hi all. i was using github actions to deploy flows and create/update blocks on prefect (v2) it was running well like a week back but today (without any changes to api keys or the workflow definition) the step where prefect config is set has started failing . attaching screenshot for the same.

Hello everyone! I am trying to deploy prefect orion server behind an nginx reverse proxy, at a different location than

/

, for instance

/prefect/

. I am having an issue with loading the UI when I access

<http://localhost/prefect/>

as the UI cannot load the corresponding assets, because it ignores the location prefix when doing the requests, for instance:

<http://localhost/assets/index.1a4d60fb.js>

is available at

<http://localhost/prefect/assets/index.1a4d60fb.js>

. Is there a way to specify a prefix or some sort of redirect URL for the UI to use when doing requests?

Hi all, I’m running into an issue with Prefect GQL where I get the following error message

{
  "errors": [
    {
      "path": [
        "flow_run",
        0,
        "id"
      ],
      "message": "Cannot return null for non-nullable field flow_run.id.",
      "extensions": {
        "code": "INTERNAL_SERVER_ERROR"
      }
    }
  ],
  "data": null
}

When I add the following “where” filter it seems to work (this would suggest I have some flows with null ids?), however when running a query to find flows with null ids nothing is returned.

{
  "where": {
    "flow": {
      "id": {
        "_is_null": false
      }
    }
  }
}

Please let me know if you have any ideas about why this may happen, thanks!

Hello How can we deploy multiple agents on the same kubernetes cluster, and the same namespace ? Apparently, with using the helm charts, we can just deploy one agent, if I try to deploy another, I get some problems with the role binding / role itself (as I will be overwriting the current declaration by binding it to another agent).

hello community, hello prefect team, Is there any new rate limiting for the cloud API in place? when using prefect 2 cloud, i'm recently facing a lot of flow crashes with the error message:

Crash detected! Request to <https://api.prefect.cloud/api/accounts/[...]/workspaces/[...]/task_runs/> failed. 
...
RuntimeError: The connection pool was closed while 65 HTTP requests/responses were still in-flight.

this is most likely not related to local network issues as it is reproducible across different networks. when using a local orion server instead, the flow succeeds without any issues and a couple of days ago, the flow ran successfully on prefect cloud. I am submitting ~100 parallel tasks using the RayTaskRunner with a local cluster. Any help is much appreciated. python 3.8.10 prefect 2.4.5 prefect-ray 0.2.0.post2

Hi, I'm running prefect (orion and agent) on EKS with

kubernetesJob

on EKS orion, agent, and prefect cli are all

2.4.5

When I submit a simple task I get the following error in the agent logs:

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"jobs.batch is forbidden: User \"system:anonymous\" cannot create resource \"jobs\" in API group \"batch\" in the namespace \"prefect\"","reason":"Forbidden","details":{"group":"batch","kind":"jobs"},"code":403}

Seems that for some reason the service account called

agent

isn't being observed?

system:anonymous

is being used instead. The

role

and

roleBinding

both match what is in the helm chart so permissions should be fine.

$ kgp agent-5fbdcf4bbb-zrrg7 -o yaml | grep serviceAccount
  serviceAccount: agent
  serviceAccountName: agent

I was not having these problems before upgrading from 2.4.0 -> 2.4.5 😞. I created a new kubernetesjob block for 2.4.5 to silence an error