I have a prefect (version 2.10.3) flow structured like the following example, where the

cleanup

task must happen after all the

my_task

tasks have completed. To make that happen, I'm passing the results of the mapped

my_task

tasks, even though

cleanup

doesn't do anything with them.

@flow
def my_flow():
    results = my_task.map([1,2,3])
    cleanup(results)

@task
def my_task(x, dummy):
    # ...do whatever
    return value

@task
def cleanup(dummy):
    # ...do something essential (without using `dummy`)

Sometimes, one of the

my_task

tasks fails, and the

cleanup

task stays stuck in

NotReady

state, according to the Prefect Cloud web UI. Is this expected behaviour? If so, what's the recommended way to ensure that

cleanup

always runs after all the

my_task

tasks have finished, whether they succeeded or not?

Hello @Prefect, I am trying to turn off logging to cloud through settings? Do I need to need to create environment specific profile to toggle these settings. Could you please throw some light on this? https://docs.prefect.io/latest/api-ref/prefect/settings/#prefect.settings.PREFECT_LOGGING_TO_API_ENABLED

my flow has an argument

x

. when I do a

quick run

there is a UI prompt to fill in

x

. when I set a cron schedule, I see no such prompt. what value is it using for

x

, just

None

? how do I set this per-deployment? even better, how do I set it relative to the scheduled run time, like how Airflow has

{{ ds_nodash }}

for a daily run?

I know I can get the actual time the flow was entered with

prefect.context.get_run_context().start_time

, but that seems a little dangerous if the run is late; I'd rather pull the scheduled time if possible

not sure if this is the real cause, but plausible fix to my issue https://prefect-community.slack.com/archives/CM28LL405/p1681244197960319?thread_ts=1681235909.744809&cid=CM28LL405 (which is still occurring) proposed here https://github.com/PrefectHQ/prefect-aws/pull/240 . not actually sure what the best way to test this is without building a whole new image from my fork so appreciate any help there.

Hey everyone! Prefect Cloud Customer getting the following error when trying to execute a flow on AWS ECS Fargate

Downloading flow code from storage at 'orchestration/main-orchestration-flow'
03:31:38 PM
prefect.flow_runs

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 879, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1016, in get_code
  File "<frozen importlib._bootstrap_external>", line 1073, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/main_orchestration_flow.py'

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 276, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/client/utilities.py", line 40, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/deployments.py", line 217, in load_flow_from_flow_run
    flow = await run_sync_in_worker_thread(load_flow_from_entrypoint, str(import_path))
  File "/usr/local/lib/python3.10/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.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 "/usr/local/lib/python3.10/site-packages/prefect/flows.py", line 809, in load_flow_from_entrypoint
    flow = import_object(entrypoint)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/importtools.py", line 201, in import_object
    module = load_script_as_module(script_path)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/importtools.py", line 164, in load_script_as_module
    raise ScriptError(user_exc=exc, path=path) from exc
prefect.exceptions.ScriptError: Script at 'main_orchestration_flow.py' encountered an exception: FileNotFoundError(2, 'No such file or directory')

I saw that a bug was fixed in 2.10.3, but this flow is running on that version. The files are in the remote storage.

cross posting this question: im using the helm server and work charts and i've isolated the pods with nodeselector as usual, but is there a way to isolate the job containers that spin up? they seem to create on any node they want by default. how can i control what nodes these spawn on?

Hello @Prefect I am trying to create prefect service accounts though rest api. I am spinning the docker container . It is throwing me the *Value Error("prefect.api.cloud) does not appear to be an IPV4 or IPV6 address". Can someone please throw light on this"?

My flow subflow is stuck in running state, but all tasks are completed.. Is there a timeout on flow?

Hi, I am Renilton from Brazil (Logcomex company). I am starting to use prefect 2.0 cloud. I would like to know if the infrastructure could be similar of prefect 1.0 cloud, like: Can I create a docker image, installing prefect lib and python lib, doing login with token-agent in prefect service trought container, running prefect agent in container, etc etc?

Hi! I have a flow that runs without issues locally but does not start when running on cloud, it just says status

Late

. How can I debug what the problem is?

I'd like to set up slack notifications, following these instructions, but there is no "Notifications" tab in our workspace.

Getting

Submission failed. KeyError: 'env'

when attempting a "Hello World" prefect flow on ECS Using the prefect-aws guide and using this invocation to build a deployment:

prefect deployment build run.py:main \
    -n jack-test-1 \
    -ib ecs-task/ecs-block \
    -sb s3/flow-storage-block \
    --pool some-pool \
    --override env.EXTRA_PIP_PACKAGES=prefect-aws

Hi , #CM28LL405

Hi , when migrating from Prefect1 to Prefect2 I encountered a problem. The local agent always failed with following error msg:

___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from work pool 'default-agent-pool'...

Failed the last 3 attempts.  Please check your environment and configuration.
Examples of recent errors:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/h2/connection.py", line 224, in
process_input
    func, target_state = self._transitions[(self.state, input_)]
KeyError: (<ConnectionState.CLOSED: 3>, <ConnectionInputs.SEND_HEADERS: 0>)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/http2.py", line
116, in handle_async_request
    await self._send_request_headers(request=request, stream_id=stream_id)
  File "/usr/local/lib/python3.10/dist-packages/httpcore/_async/http2.py", line
213, in _send_request_headers
    self._h2_state.send_headers(stream_id, headers, end_stream=end_stream)
  File "/usr/local/lib/python3.10/dist-packages/h2/connection.py", line 766, in
send_headers
    self.state_machine.process_input(ConnectionInputs.SEND_HEADERS)
  File "/usr/local/lib/python3.10/dist-packages/h2/connection.py", line 228, in
process_input
    raise ProtocolError(
h2.exceptions.ProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in
state ConnectionState.CLOSED

The local server info as below:

Ubuntu 22.04
$ prefect version
Version:             2.10.3
API version:         0.8.4
Python version:      3.10.6
Git commit:          f9ddd259
Built:               Tue, Apr 11, 2023 11:55 AM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         cloud

It seems a network issue, but never happened on Prefect 1. Do you have any thoughts?

Is it normal for files to be written to S3 when a deployment is built? Or are they supposed to be written there when the deployment is applied? This is what I'm seeing:

$ prefect deployment build run.py:main         -n jack-test-3         -ib ecs-task/ecs-block         -sb s3/flow-storage-block         --pool some-pool   --override env.EXTRA_PIP_PACKAGES='prefect-aws s3fs'
Found flow 'main'
Deployment YAML created at 'V:\some-dir\ecs-test-3\main-deployment.yaml'.
Successfully uploaded 2 files to <s3://some-bucket/some-path/>

When setting up a flow to run on ECS, is it the

task_role_arn

or the

execution_role_arn

that needs read access to the S3 bucket where the flow is deployed?

Hey all, I know there are several different ways to utilize prefect cloud on AWS, but I am trying to find the best solution for my companies use case. We are going to be utilizing prefect cloud for both our data transformation piece servicing many customers along with data ingestion from various api's. First question: As far as agents go, I was wondering the limitations behind agents taking in tasks. In our end objective we fear the workload agents may receive. From what I understand the relationship of agents to workpools is one to many, but workpools can only have one agent. I was wondering if you could enlighten me on some usecases you have seen in the past? Second question: Agents are used to look for deployments that need to be run that exist in a specific work queue. In this case, is there a feature or will there be in the future, for when a job needs to be run the agent will receive the flow run request and spin up infrastructure on the fly to accommodate the workload? I understand the uses of ECS, but it seems like there can be another way maybe utilizing IAM roles that have access to different resources. Third question: Do you have a specific article I can follow for getting started with prefect cloud in AWS. I have been able to launch the agent and utilize an EC2 for compute, but I want more visibility. I would thoroughly enjoy maybe some talk from people who have implemented prefect cloud utilizing EKS or ECS. I am a little lost and when I have followed the documentation I have found online it has led to no success. Thanks a bunch and I look forward to hearing back from you!

Recommendations for how to make a backup of an infrastructure block? So we can restore to a known working state if it gets accidentally changed.

Hi! We are using Prefect Cloud on GCP set up accoring to @Anna M Geller s blog post here https://medium.com/the-prefect-blog/gcp-and-prefect-cloud-from-docker-container-to-cloud-vm-on-google-compute-engine-2dffa026d16b How ever, after a flow running for one hour, Cloud Run times out, and the job stops. Questions: 1. Is there anyway around this 1hr timeout? 2. When it times out the flow gets status = Successful, which its not correct. Any way to fix that?

Hello everyone ! Is it possible to deploy the Prefect agent on a lambda with SQS ? (sorry if it's the wrong channel for questions) Or any other serverless way? The idea is to deploy the most cost-effective solution possible (not have a 24/7 runnning instance/container for the agent). Thank you all !

This message was deleted.

Hey all, I've been using Prefect on an AWS VM for various automations at my company for 6+ months. It's working well, but we're hitting a point with the number of contributors where it's common for someone to be waiting their turn to remote into the VM for make adjustments or publish a new flow. Prefect Cloud seems like the obvious option, but our head of IT says he won't consider any new systems that can't use Azure AD as an identity provider. From what I can tell, Prefect Cloud supports GitHub and Gmail, but not Azure. Is there any workaround here?

Hi, I ’ve started a worker last night and this morning I ve just found it crashed because of this

return fn(*args, **kwargs)
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/prefect/utilities/asyncutils.py", line 260, in coroutine_wrapper
    return call()
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/prefect/_internal/concurrency/calls.py", line 245, in __call__
    return self.result()
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/prefect/_internal/concurrency/calls.py", line 173, in result
    return self.future.result(timeout=timeout)
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 428, in result
    return self.__get_result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/prefect/_internal/concurrency/calls.py", line 218, in _run_async
    result = await coro
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/prefect/cli/worker.py", line 142, in start
    started_event = await worker._emit_worker_started_event()
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 662, in __aexit__
    raise exceptions[0]
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 702, in _run_wrapped_task
    await coro
  File "/home/sgonzalez/venv/lib/python3.7/site-packages/prefect/utilities/services.py", line 104, in critical_service_loop
    raise RuntimeError("Service exceeded error threshold.")
RuntimeError: Service exceeded error threshold.
An exception occurred.

And the thing is that, no flow has been running last night. Do you know what is the cause of the crash of the worker? Prefect version

2.10.5

.

Hello Everyone, In the word doc attached is my inquiry including screenshots of my progress. It revolves around getting pods to execute flowruns properly in an eks cluster.

Hello everyone, Facing an issue in launching agent in the Kubernetes deployment. I have created the work pool and work queue in my prefect UI. Able to launch agent in my local machine by pointing to the a work pool and work queue that agent is getting started perfectly.

prefect agent start -p <WORK_POOL> --api <WORK_SPACE_API>

But when i try to do the same in the Kubernetes deployment it is not working.

Hello, I was wondering if anyone had any insight to my inquiry above? Thanks! https://prefect-community.slack.com/archives/CM28LL405/p1682086945603549