Hello, I have a prefect agent hosted in my own AWS environement that seems to be suddenly switching to an unhealthy state, where checking the agent logs seems to throw an errror saying that the ecs role I've assigned to the agent does not have permission to get the log events for

arn:logs:{region}:{id}:log-group:prefect:log-stream:alpha6-matar/prefect/{log id}

during ecs flow runs. I would understand adding this permission during the setup, but I'm confused as to why it be working fine last week and suddenly experience hiccups (including a brief moment of uptime) today.

Hi, I wonder if anyone has insights into an intermittent issue I've been encountering. I'm scheduling flow runs on ECS Fargate. The majority of the time they run successfully, however they occasionally fail with this error:

State message: Submission failed. RuntimeError: Timed out after 120.73846864700317s while watching task for status {until_status or 'STOPPED'}

Is this a known issue? Any ideas how to prevent this, or at least retry the flow run in the case of an intermittent error like this?

Getting a lot of these lately and not sure if there is anything I can do about it:

Crash detected! Execution was interrupted by an unexpected exception: PrefectHTTPStatusError: Server error '500 Internal Server Error' for url '<https://api.prefect.cloud/api/accounts/foo_account/workspaces/foo_workspace/flow_runs/foo_flow_run_id/set_state>'
Response: {'exception_message': 'Internal Server Error'}
For more information check: <https://httpstatuses.com/500>

^We're pretty much only getting that rn?

We are investigating higher latency / errors in the Cloud 2.0 AI. The status page has been updated.

Hey all. Hoping someone could point me in the right direction. Trying to use

run_deployment

to kick off one of our deployed flows after an event is triggered within a lambda. I am currently getting a 401 error. I've already added the

PREFECT__CLOUD__API_KEY

. Anything else I need to do? Any documentation would help too. Thank you

Weird cache expiration behavior. I’m using

@task(cache_key_fn=task_input_hash)

so the Flow() will not execute the code inside tasks that have been completed before. I did not change the inputs to the task signature, but I did add a simple

time.sleep(10)

into my task code. I went away for 1-2 minutes and re-ran the task expecting a Cached(Completed()) status, but my task code ran again. Documentation says that unless specified, the cache never expires. So I wonder if changing the task code itself (without changing the task inputs) changes the hash?

Hello @Prefect Team - I am trying to run a simple 'Hello World' Prefect Flow on AKS using k8s infra block but looks like its throwing an exception when retrieving code because of permission related issue. Also, my flow code is baked into the docker image.

Are there any known caveats to using standard

boto3

in Prefect flows? I have a task that I'm iterating over an input, and if I map it or

.submit() -> .result()

in a for loop for each item, certain boto calls take upwards of 3 minutes to resolve on ECS (not locally).

Hi @Prefect Team , I am using Prefect2.0 ,Getting error while using gitlab as storage :Downloading flow code from storage at '.' . Flow could not be retrieved from deployment. NotImplementedError. Can anyone help me to understand resolve the issue

Hey folks, prefect 2.0 cloud user here. Yesterday evening (CET) prefect flows started to fail with ``Server error '500 Internal Server Error'`` and after runs stopped failing ALL deployments had a pending run - and the runs stayed in pending forever. They stayed in pending even though we have an automation which says "kill all runs staying in pending for longer than 1 hour". I can't provide any logs, as there were none, I guess this was related to the cloud outage described above. That's fine for me, however, I have a question related to that: Is there a way to prevent this "staying in pending forever"? We have some sensitive pipelines with concurrency 1 - so if one of the runs is in pending - all the others will wait forever.

HI P Team, I have some issues with caching. I was told that "Failed tasks are not cached" . I think that a cache record is still made on the database even when the task run fails. My Case: Caching all responses from the API but the API isnt 100% reliable hence sometimes causes

requests.exceptions.ReadTimeout

Prefect attempts to cache this exception but Exceptions arent JSON serializable. So

Object of type \'ReadTimeout\' is not JSON serializable\n')

is raised. But Upon trying to retrigger the flow, this exception is raised

prefect.exceptions.MissingResult: The result was not persisted and is no longer available.

I guess there is a cache record on the database but the cache key is not available in the storage location because it failed while dumping to the storage.

async tasks give us the same error since we upgraded from 2.8.2 to 2.9

Hi community/@Prefect I’ve been struggling with parallel subflow runs. Using asyncio trick is allowing them to run in parallel, but still on the same agent that picked up the main flow. Is there a way to add the subflows to a work queue, so they can be picked up by several agents?

Hey all! Can anybody help with a github storage block? When I run the flow, I keep getting the message

Flow could not be retrieved from deployment

with a stack trace that leads to a

NotImplementedError

. I have tried several variations on the --path argument, tried a public repo instead of a private - nothing works! Please help! flow-code:

import sys
import prefect
from prefect import flow, task, get_run_logger
from utilities import AN_IMPORTED_MESSAGE


@task
def log_task(name):
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello %s!", name)
    <http://logger.info|logger.info>("Prefect Version = %s 🚀", prefect.__version__)
    logger.debug(AN_IMPORTED_MESSAGE)


@flow()
def log_flow(name: str):
    log_task(name)


if __name__ == "__main__":
    name = sys.argv[0]
    log_flow(name)

deployment CLI command:

prefect deployment build ./src/log_flow.py:log_flow -n log-simple-test -q test -sb github-repository/test-public-repo --apply

log_flow-deployment.yaml:

###
### A complete description of a Prefect Deployment for flow 'log-flow'
###
name: log-simple-test
description: null
version: 0309032a075c5be33c8c074988fd6e41
# The work queue that will handle this deployment's runs
work_queue_name: test
work_pool_name: default-agent-pool
tags: []
parameters: {}
schedule: null
is_schedule_active: true
infra_overrides: {}

###
### DO NOT EDIT BELOW THIS LINE
###
flow_name: log-flow
manifest_path: null
infrastructure:
  type: process
  env: {}
  labels: {}
  name: null
  command: null
  stream_output: true
  working_dir: null
  _block_document_id: ce44b1b5-b0a4-4034-a724-b4660da17555
  _block_document_name: anonymous-fa985916-1aed-4f2b-89f2-5f5e2c278ae0
  _is_anonymous: true
  block_type_slug: process
  _block_type_slug: process
storage:
  repository_url: <https://github.com/zachloertscher/test-public-repo>
  reference: main
  credentials: null
  _block_document_id: a2fbc663-3d11-4551-9350-49eec1a3de54
  _block_document_name: test-public-github-repo
  _is_anonymous: false
  block_type_slug: github-repository
  _block_type_slug: github-repository
path: /Users/zachloertscher/test-public-github-repo
entrypoint: src\log_flow.py:log_flow
parameter_openapi_schema:
  title: Parameters
  type: object
  properties:
    name:
      title: name
      position: 0
      type: string
  required:
  - name
  definitions: null
timestamp: '2023-04-04T13:57:58.625897+00:00'

Stack trace:

09:58:36.212 | INFO    | prefect.agent - Submitting flow run '0b7f9e6d-bbcd-4b21-8c41-12d735d31816'
09:58:36.874 | INFO    | prefect.infrastructure.process - Opening process 'dark-curassow'...
09:58:36.978 | INFO    | prefect.agent - Completed submission of flow run '0b7f9e6d-bbcd-4b21-8c41-12d735d31816'
C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\runpy.py:126: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour
  warn(RuntimeWarning(msg))
09:58:41.105 | INFO    | Flow run 'dark-curassow' - Downloading flow code from storage at '/Users/zachloertscher/test-public-github-repo'
09:58:41.113 | ERROR   | Flow run 'dark-curassow' - Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\engine.py", line 277, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\client\utilities.py", line 40, in with_injected_client
    return await fn(*args, **kwargs)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\deployments.py", line 194, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=deployment.path, local_path=".")
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect_github\repository.py", line 146, in get_directory
    process = await run_process(cmd, stream_output=(out_stream, err_stream))
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\utilities\processutils.py", line 258, in run_process
    async with open_process(
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\contextlib.py", line 199, in __aenter__     
    return await anext(self.gen)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\utilities\processutils.py", line 200, in open_process
    process = await _open_anyio_process(command, **kwargs)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\utilities\processutils.py", line 162, in _open_anyio_process
    process = await asyncio.create_subprocess_shell(
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\asyncio\subprocess.py", line 205, in create_subprocess_shell
    transport, protocol = await loop.subprocess_shell(
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\asyncio\base_events.py", line 1634, in subprocess_shell
    transport = await self._make_subprocess_transport(
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\asyncio\base_events.py", line 498, in _make_subprocess_transport
    raise NotImplementedError
NotImplementedError
09:58:41.292 | ERROR   | prefect.engine - Engine execution of flow run '0b7f9e6d-bbcd-4b21-8c41-12d735d31816' exited with unexpected exception
Traceback (most recent call last):
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\engine.py", line 2131, in <module>
    enter_flow_run_engine_from_subprocess(flow_run_id)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\engine.py", line 202, in enter_flow_run_engine_from_subprocess
    ).result()
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\client\schemas.py", line 107, in result
    return get_state_result(self, raise_on_failure=raise_on_failure, fetch=fetch)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\states.py", line 76, in get_state_result
    return _get_state_result(state, raise_on_failure=raise_on_failure)
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\utilities\asyncutils.py", line 260, in coroutine_wrapper
    return call()
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\_internal\concurrency\calls.py", line 245, in __call__
    return self.result()
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\_internal\concurrency\calls.py", line 173, in result
    return self.future.result(timeout=timeout)
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\concurrent\futures\_base.py", line 439, in result
    return self.__get_result()
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\concurrent\futures\_base.py", line 391, in __get_result
    raise self._exception
  File "C:\Users\Zachary.Loertscher\repos\data_warehouse_pipeline\.venv\lib\site-packages\prefect\_internal\concurrency\call    transport = await self._make_subprocess_transport(
  File "C:\Users\Zachary.Loertscher\AppData\Local\Programs\Python\Python310\lib\asyncio\base_events.py", line 498, in _make_subprocess_transport    raise NotImplementedError
NotImplementedError
09:58:41.948 | ERROR   | prefect.infrastructure.process - Process 'dark-curassow' exited with status code: 1

My repo: https://github.com/zachloertscher/test-public-repo I've been at this for hours - any help would be so very much appreciated!

I need to use

set PREFECT__CLOUD__USE_LOCAL_SECRETS=false

in order to access databases while running things in Anaconda terminal. How can I replicate this when running/testing code in a VS Code interactive window?

Hi! I am wondering what the best practice is for using Docker with Prefect. I am currently using a VM on Azure as a Prefect Agent, and then I use DockerContainer as infrastructure. Would it not be better to run the agent in a dockercontainer directly, and use regular python process as infrastructure? Now there is a bit of setup to do to enable running the Prefect agent on a VM (install python/prefect etc.)

Hi! I am confused about best practice for deploying a Docker container on ECS with Infrastructure blocks. My agent will be running on a long-standing ECS cluster. Do I only need a Docker infrastructure block for this? Or do I need to use an ECS task infrastructure block? Or some combination of both? Any insights or resources would be super appreciated!

Hello! what would be the recommended way to define a flow that runs on a schedule and of which we only want 1 instance running at any given time? This would be for the cases where a first flow run takes longer to run than the deployment schedule frequency. Is there a better approach than creating some sort of marker in storage that flow runs check at the beginning?

(copying this question from #C04E1CKG456 per request from @Will Raphaelson) Is there any way to get the parent flow of a subflow? I just started using subflows, and

{{ deployment.name }}

is blank when the flow is a subflow - ideally in the automation whisker template I could check if a flow was really a subflow, and then get the subflow name and the parent deployment name. Any tips? I’ve been fiddling with

get_run_context

and

prefect.runtime

but I don’t see anything that can be used to 1) check if a flow is really a subflow, and 2) if a flow is a subflow, get the parent deployment name Clearly there is a way to do it since the Prefect UI has this logic as i toggle through pages, but I don’t see a clear way to get that same information with the prefect python library

Hello. I have a flow that calls 3 other subflows. These subflows can enter

Cancelled

state in certain conditions (expected behavior). When any of them enters

Cancelled

state and the rest complete, the main flow also enters

Canceled

state. How can I make the main flow succeed? This is my code

@flow()
def main_flow():
  subflows = (subflow1, subflow2, subflow3)  
    for subflow in subflows:
      try:
        subflow()
      except CancelledRun:
        continue

When the first flow cancels, I get something like this

Finished in state Cancelled('1/3 states cancelled.')

and the main flow state becomes Cancelled too. Thanks

Hi Team, when I try to use

Deployment.build_from_flow

from

prefect.deployments

to deploy flow to Prefect cloud, the process is stuck without erroring out and yet didn’t give any verbose status/progress. I checked the code was uploaded to S3 bucket but the deployment never happened on the Prefect cloud. Do you know what might be the root cause? and is it possible to add more verbose loggings for the

Deployment.build_from_flow

method?

Hi everybody, I am looking to implement a job outside of Databricks that can pull data from a remote API periodically, while keeping track of the next timestamp to use for each request, ie.

max(timestamp) + 1

in each returned response. Is prefect a good tool to use? I have been using databricks exclusively so this will be the first tool added to my "data stack", and it looks like I will need to move my scheduled job definitions out of databricks, and into prefect, with a step that invokes the old databricks task

This is probably a result of being unfamiliar with Prefect's verbology... but I'm having a hard time figuring out how to share state between flow runs. I am building a flow that will need store a "cursor"-like value between runs

Hi. In Prefect 2 I use context manager instead of ResourceManager available in Prefect 1. It works fine except for the case when the flow was cancelled. After flow is cancelled my resource is still running as finally block wasn't executed. Do you have any ideas how to shutdown the resource in case of cancel action? Thanks.

Hi all! Is it possible to run subflows in parallel?

Hello. I am using Prefect 2.8.6. I am running ECS task within a flow using the ECS Task block. Even if an application error occurs within the ECS task, the error is not recognized by the flow, and the flow terminates as a success. Is there any way to detect errors in ECS tasks executed in the flow other than by determining them from the

status_code

of the

ECSTaskResult

?

from prefect_aws.ecs import ECSTask
ecs_task_block = ECSTask.load("my-ecs-task-block")
ecs_task_result = ecs_task_block.run()
print(ecs_task_result.exit_code) # 1 (application error)

Hi, I'm trying to return an instance of this SourceRead class from a task, but I'm running into some Pydantic weirdness.

Good Morning prefect. Is it possible to get the latest deployment_id from the API? We are kicking off a flow run externally and are trying to just use the API and not the client. Trying to sift through the API docs and I don't see a way. Appreciate the help!

Can I start a subflow and leave it running after the parent flow ends? Basically I have a job that has a component that needs to be done sequentially (but only takes 15 seconds) but that sequential component can trigger another component that can be done concurrently (but will take 45 seconds).