What does it mean when there are “Late submittable runs”? Our agent was successfully registered and labelled with the right labels but the flows are not running.

Hello, i'm running into an issue were i have tasks that are skipped when none of their upstream tasks fail. When looking at the execution schema in the dashboard, i see that the task in charge of loading secrets like api key has been skipped because the first task that needed it has been skipped (this one is an intended behaviour). I suppose that my last task got skipped because the secret it needs as an argument has been previously skipped. Am i right ? if so how can i "force" the task to get the secret ? Here is the schema :

Hello! We're running into the same

no heartbeat detected

error that I've seen a few other folks flag. I read through the discourse documentation and the FAQ. I don't think it's a memory issue because a much larger Flow is running from the same agent + Google Cloud VM without any issues. I can't seem to use the proposed solutions to configure heartbeats to use threads because I'm realizing that my Flows do not use:

*from* prefect.run_configs

So, I'm not using Universal Run or ECSRun. I'm guessing the setup of my flow might pre-date that library. Is that possible? Should I import that library if my Flows have been running fine without it? I should mention that this heartbeat issue only started to crop up a few days ago.

Hey Community, Having an issue with the Apollo service in an EKS cluster. It works pretty well, until ~500-1000 mapped tasks are trying to update their task_run_state. The problem is that if even one fails, it gets stuck in a pending state and the flow just hangs. Eventually, the prefect-job pod exits and things continue to go sideways from there.

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1922, in set_task_run_state
    result = self.graphql(
...
    raise ConnectionError(e, request=request)
requests.exceptions.ConnectionError: HTTPConnectionPool(host='prefecthq-apollo.prefect', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f12486259d0>: Failed to establish a new connection: [Errno -2] Name or service not known'))

This error would seem to indicate that the service can't even be resolved, but this DNS name is perfectly resolvable from the dask pods, and many tasks succeed before a couple fail. I see signs of distress from the apollo service in the form of:

│ BadRequestError: request aborted                                                                                                                                                                                                           │
│     at IncomingMessage.onAborted (/apollo/node_modules/raw-body/index.js:231:10)                                                                                                                                                           │
│     at IncomingMessage.emit (events.js:315:20)                                                                                                                                                                                             │
│     at abortIncoming (_http_server.js:561:9)                                                                                                                                                                                               │
│     at socketOnClose (_http_server.js:554:3)                                                                                                                                                                                               │
│     at Socket.emit (events.js:327:22)                                                                                                                                                                                                      │
│     at TCP.<anonymous> (net.js:673:12)

repeated pretty much as many times as the dask pod tries it. Light research on this case points to the service being overwhelmed. Is this case covered by the request retrier here? https://github.com/PrefectHQ/prefect/blob/a2041c7ff1a619e611f614ed3394fccd05bb2005/src/prefect/client/client.py#L633 If not, what's the best way to handle this case? Any configuration changes I could be making the Apollo pod?

Hi, I’m currently running into issues trying to do a simple get request for logs from the GraphQL API. Could anyone help me identify where my mistake is to get the error below? I’ve tested the query in the Prefect front end’s Interactive API and it works just fine. I’ve also tested in a Javascript get request that that API key works fine, so this may be a requests library related mistake?

Hello all! When executing

prefect orion kubernetes-manifest | kubectl apply -f -

command,

kubernetes

queue is not created, although according to the documentation it should. Has anyone had similar?

Hey guys any working example of how to upload csv files from local machine to S3 using awswrangler? Of course prefect task

hi guys, when i hit run on a prefect ui on the cloud, its creating a separate run rather than just restarting the run i had just started. • is there a way to run it just once? • Another thing is, if i run a deployment file with many deployments, like 30, its kind of painful to manually go to the UI and hit run on all of them. there must be some ability in the deployment spec to ensure a file thats not on a schedule gets run immediately upon deployment. • i want the ability to constantly check if a task is running and if its not, to initiate a run. a signal or whatever. • ability to email my team and I in case something crashes on the prefect side that was not picked up on the linux side

Starting in 5 minutes! See you on Twitch! https://prefect-community.slack.com/archives/CL09KU1K7/p1648046010730119

I’m like 99% certain that this is the case but I’m just looking to confirm: when I have a task that creates a piece of mutable data, say a dictionary, and I pass it downstream to two different tasks, both tasks get a reference to the original dict right? So if one task edits it there risk for conflict. ex:

with Flow('foo') as f:
    a = task(lambda x: {'key': x})('value')
    b = task(lambda y: y.pop('key'))(a)
    c = task(lambda z: z.get('key'))(a)

Is it possible to configurate Orion to use `KubernetesFlowRunner`+

DaskTaskRunner

that use

dask_cloudprovider.aws.FargateCluster

? There is no logs in Orion UI except

'task-b610b3b0-0' - Crash detected! Execution was interrupted by an unexpected exception.

and

RuntimeError: IOLoop is closed

but it's possible to see it in CloudWatch console. It contains the following error:

Exception: 'RuntimeError("Cannot orchestrate task run \'9987d480-fad1-433a-9749-077a63fbdc0a\'. Failed to connect to API at <http://orion:4200/api/>.")'

. Note: Orion is deployed on Kubernetes via

orion kubernetes-manifest

command.

Hey Prefect team, I have an urgent question regarding Prefect Cloud. Since switching from the legacy payment tier and hitting the 20k free cap all our tasks are halted UsageExceeded. We have a credit card on file but nothing is going through. Please help as we've ground to a standstill.

Hey Prefect team! I was wondering if there's documentation or a github repo that I can look at for integrating Singer taps with Prefect? What I would like to do is run Singer taps in a prefect flow. I did some research and I see Singer tap integrations with Airflow and Dagster so I assume Prefect has the capabilities too but I can't seem to figure it out 😅

If I have a flow like :

with Flow("flow", run_config=RUN_CONFIG, storage=STORAGE) as flow:
  json_data = get_json_data(
    url, query, headers,
    task_args={"name": "Getting Flow Data"}
  )

How can I then access json_data as a python object, rather than a FunctionTask object?

Hi all, was trying to rename my Team but unable to so deleted it, but now not able to add a new one? Is this a bug, will this be a big issue?

What is the difference between

StartFlowRun

vs

create_flow_run

? They seem to be slightly different in the documentation but has the same end result

Hello all. I've run across an issue that I just can't seem to figure out and I feel like it's probably something really simple. I've structured my project like so.

my_module/
  flows/
    my_flow.py
  tasks/
  lib/
     tasks/
       shared_task.py

So in

my_flow.py

I am importing from

my_module.lib.tasks.shared_task.py

where i've created a reusable Task class. But I am unable to register this flow using the prefect command

prefect register --project Test -m <http://my_module.flows.my|my_module.flows.my>_flow

I've tried many different combinations of the

prefect register

command and it just continues to generate a

ModuleNotFoundError: No module named my my_module

However, if I just use

flow.register()

inside of

my_flow.py

and execute it using

python my_module/flows/my_flow.py

it registers just fine.

Logs not visible on UI , how to fix this ?

Hey guys, Is there a way we can access the Postgres DB for the Prefect and query data from it. I am unable to find access the database. Any commands, default paths, or documentation would be helpful. Thanks

What is the proper way to stop local Prefect? I run 'prefect server start' as backend server and then press Ctrl+C to stop it, next time you start it Prefect throws a lot of duplicate violation errors and doesn't start properly (you can't create a project for example)

Also there 2 more strategic questions: 1. How doesn't Prefect understand that flow fails? I have a Flow with a single Task. Task calls some external module and that module throws an exception, but Task and Flow in Prefect completes successfully?

2) How to handle multithreading inside custom modules? Task runs from Prefect, but code defined in custom module. If there is a multithreading block like below, Prefect will not wait until all futures/threads are completed, but succeed task immediately and exit.

# with concurrent.futures.thread.ThreadPoolExecutor(max_workers=10,
#                                                   thread_name_prefix="flow_") as executor:
#     futures = [executor.submit(self._run, df) for df in self.dfs]
#
#     for future in as_completed(futures):
#         try:
#             future.result()
#         except Exception as exc:
#             logger.exception(exc)

Hey, is there a way to instantly force the execution of a PrefectSecret task and set it with no upstream ?

Hey. I would like to change

ephemeralStorage

in ECSRun configuration. Can I use

run_task_kwargs

to modify it? If yes - what is the format of this dict to pass

ephemeralStorage

?

Hi all I am currently trying to get k8s agent working on gcp autopilot cluster - have run into the following error;

jobs.batch is forbidden: User \"system:serviceaccount:default:default\" cannot create resource \"jobs\" in API group \"batch\" in the namespace \"prefect\"

Hi all! I had a very strange behaviour with a task being executed twice, with a few seconds interval:

16:46:32 Task 'x': Starting task run...
16:46:33 Task 'x': Finished task run for task with final state: 'TriggerFailed'
01:39:32 Task 'x': Starting task run...
01:39:46 Task 'x': Starting task run...

Don't know if it could be linked, but I had to retry the parent task that failed (hence the "TriggerFailed" that you can see above). I'm running using a k8s executor, with core version 0.14.20. Any idea what could have happened?

Hello Prefecters! What's up? I tried to create a schedule flow using:

from prefect import flow
from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule
from datetime import timedelta


@flow(name="TestingFlow")
async def testing_flow(name="raimundo"):
    try:
        print(f"Hello {name}!")
    except Exception as e:
        print(e)


DeploymentSpec(
    flow=testing_flow,
    name="hw-30s",
    schedule=IntervalSchedule(interval=timedelta(seconds=20)),
    tags=["rai", "20s"],
)

when I run

prefect deployment create my_file.py

, that creates it correctly, but the tasks don't run. Anyone help me please 💙

Hi Prefect team! I have a couple questions about Orion capabilities within our production environment. Currently we are running Prefect core with prefect server on a local production server and agents on separate vm servers. This makes it really easy to run automated capabilities, process flows quickly, etc. We really like how flows can be defined and encapsulated in docker with dependencies and shared within our architecture. We don’t like how Flow of Flows is statically defined within Prefect Core, but I believe this will be solved with prefect Orion. • Am I going to be able to set Orion up in a similar architecture? Can I have Orion set up on a local server that employees can access via web ui and run capabilities? Will it be able to store flow data and logs? • I was wondering what docker storage is going to look like within a production infrastructure. Is there going to be support for docker storage in gitlab like with prefect core? • I really like Orion’s developer simplicity and how easy it is to create flow of flows. Is there going to be a way for flow of flows to be dynamically created in the UI? As in, like some sort of functionality to drag and drop flows in a particular order or check box which flows you want to run. Thanks for all your help!

Something I am trying to understand with prefect cloud 2.0 is compute part of the service? So wondering if prefect will handle infra and scaling

Hi prefecters, I have a question about prefect core (no Prefect Server, no Prefect Cloud). I'm trying to understand Flow concurrency. Is the best way to handle a stream of input data to run multiple flows (in their own thread) as the data comes in?