Hey everyone 👋 when trying to update a block using the API for prefect cloud I am getting a vague

Internal Server Error

as the response. Was not sure if this was a known issue

1.4 Question about

slack_notifier

from

from prefect.utilities.notifications

. I've been using this with great success for years on data-pipeline but I'm getting a JWT error when trying to set it up on a new prefect 1.4 flow:

Missing Authorization header in JWT authentication mode

. I have my slack url setup as an

config.toml

secret just like in my working application. Any ideas about why I'm seeing this error?

Is there a way to configure Prefect deployments so that kube pods used to execute a flow run is created in a specific subnet? I need to do this because our flow run makes third party API requests, so the pods need to be in a public subnet with all the firewall rules turned on

Hi. I enabled

Marvin AI

in our prefect account to test it out. However, I am unable to disable or turn it off? I see a setting that looks to be disabled already. I did try toggling it on and off, but we are still seeing the Marvin AI logs. There are no documentations on this anywhere. Does anyone here have any information to help us turn off this feature. Thanks

Hey, I want to make use of prefect ability to upload flows directly to s3, however the closest that I managed was a 1.0 [example](https://github.com/anna-geller/packaging-prefect-flows/blob/master/flows/s3_local_run.py) by Anna. My storage block

from prefect.filesystems import S3
import os

def s3_storage() -> None:
    block = S3(
    bucket_path="prefect-portal-us-east-1-qa",
    aws_access_key_id=os.getenv("QA_AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("QA_AWS_SECRET_ACCESS_KEY"),

).save("ecs-s3-qa", overwrite=True)

if __name__ == "__main__":
    s3_storage()

An example

from prefect_aws import AwsCredentials
from prefect_aws.ecs import ECSTask

from prefect import flow, task
from prefect.deployments.deployments import Deployment
from prefect.filesystems import S3

[...]

@flow(log_prints=True)
def cool_numbers(nums=[1, 2, 3, 5, 8, 13]): # essentially map_flow
    print_nums(nums)
    squared_nums = square_num.map(nums)
    print_nums(squared_nums)


if __name__ == "__main__":
    aws_credentials_block = AwsCredentials.load("aws-credentials-qa") # loads my aws cred
    ecs_task_block = ECSTask.load("ecs-cluster-agent-qa") # refers to the aws ecs cluster using
    s3_block = S3.load("ecs-s3-qa") # doesn't really work
    deployment = Deployment.build_from_flow(
        flow=cool_numbers,
        name="ecs-cool-numbers",
        work_queue_name="default",
        infrastructure=ecs_task_block,
        path="/",
        apply=True,
        tags=["ecs-testing"],
    )

The tl;dr is that prefect can't seem to upload to s3 and I can't execute the flow as its looking in s3.

Hi, We tried to configure prefect deployment workerpool on our ECS service but we were not able to run the task from ECS container. 1. While using the container image prefecthq/prefect:latest, the image pulling showing error message. Task is stopping CannotPullContainerError: pull image manifest has been retried 5 time(s): failed to resolve ref docker.io/prefecthq/prefect:latest: failed to do request: Head "https://registry-1.docker.io/v2/prefecthq/prefect/manifests/latest": dial tcp 3.216.34.172443 i/o timeout Please note, we tested the same scenario with docker image prefecthq/prefect:2.13.0-python3.9, the service were able to pull the image with this version. 2. After starting the service, the service got exited with exit code 2 without any error messages or logs. 3. The ECS cluster configuration consist of dedicated VPC which having private and public subnets along with NAT gateway. Could you please help us to resolve the case ? Are we missing anything on the configuration ? We followed the steps which described in the doc: https://prefecthq.github.io/prefect-aws/ecs_guide/ Kindly let us know if you need any further details from our side to replicate the case in detail cc : @Arjun J @Bianca Hoch

Hello! I have 2 questions regarding flow run cancellation. The docs mention that:

If the run does not terminate after a grace period (default of 30 seconds), the infrastructure will be killed, ensuring the flow run exits.

1. How can we change this default grace cancellation period of 30 seconds? 2. If an on _`on_cancellation`_ hook takes more time than this to complete what will happen? Will it get killed?

Hi i’m having issues running an ecs task from a worker pool in an ECS service.

@flow
def my_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello from ECS!!")

My flow configuration looks like:

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

###
### DO NOT EDIT BELOW THIS LINE
###
flow_name: my-flow
manifest_path: null
infrastructure:
  type: ecs-task
  env: {}
  labels: {}
  name: null
  command: null
  aws_credentials:
    aws_access_key_id: null
    aws_secret_access_key: null
    aws_session_token: null
    profile_name: null
    region_name: null
    aws_client_parameters:
      api_version: null
      use_ssl: true
      verify: true
      verify_cert_path: null
      endpoint_url: null
      config: null
    block_type_slug: aws-credentials
  task_definition_arn: null
  task_definition: null
  family: null
  image: prefecthq/prefect:2-python3.10
  auto_deregister_task_definition: true
  cpu: null
  memory: null
  execution_role_arn: some-execution-role-arn
  configure_cloudwatch_logs: true
  cloudwatch_logs_options: {}
  stream_output: null
  launch_type: FARGATE
  vpc_id: some-vpc-id
  cluster: cluster-worker-arn
  task_role_arn: null
  task_customizations:
  - op: add
    path: /networkConfiguration/awsvpcConfiguration/securityGroups
    value:
    - some_security_group
  task_start_timeout_seconds: 120
  task_watch_poll_interval: 5.0
  _block_document_id: 99c69452-38cb-4ce0-a0f9-65035b2db175
  _block_document_name: default-ecs-job
  _is_anonymous: false
  block_type_slug: ecs-task
  _block_type_slug: ecs-task
storage: null
path: /opt/prefect/flows
entrypoint: minerva/test.py:my_flow
parameter_openapi_schema:
  title: Parameters
  type: object
  properties: {}
  required: null
  definitions: null
timestamp: '2023-09-27T14:08:05.642381+00:00'
triggers: []
enforce_parameter_schema: null

The error I’m getting:

Failed to submit flow run '180f4415-b052-40d4-9f7c-21c1b6220835' to infrastructure.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/tenacity/_init_.py", line 382, in _call_
    result = fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect_aws/workers/ecs_worker.py", line 1524, in _create_task_run
    return ecs_client.run_task(**task_run_request)["tasks"][0]
  File "/usr/local/lib/python3.10/site-packages/botocore/client.py", line 535, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.10/site-packages/botocore/client.py", line 980, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.ClusterNotFoundException: An error occurred (ClusterNotFoundException) when calling the RunTask operation: Cluster not found.

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_aws/workers/ecs_worker.py", line 724, in _create_task_and_wait_for_start
    task = self._create_task_run(ecs_client, task_run_request)
  File "/usr/local/lib/python3.10/site-packages/tenacity/_init_.py", line 289, in wrapped_f
    return self(f, *args, **kw)
  File "/usr/local/lib/python3.10/site-packages/tenacity/_init_.py", line 379, in _call_
    do = self.iter(retry_state=retry_state)
  File "/usr/local/lib/python3.10/site-packages/tenacity/_init_.py", line 326, in iter
    raise retry_exc from fut.exception()
tenacity.RetryError: RetryError[<Future at 0x7f997b4445b0 state=finished raised ClusterNotFoundException>]

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/workers/base.py", line 843, in _submit_run_and_capture_errors
    result = await self.run(
  File "/usr/local/lib/python3.10/site-packages/prefect_aws/workers/ecs_worker.py", line 567, in run
    ) = await run_sync_in_worker_thread(
  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 33, 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 877, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 807, in run
    result = context.run(func, *args)
  File "/usr/local/lib/python3.10/site-packages/prefect_aws/workers/ecs_worker.py", line 728, in _create_task_and_wait_for_start
    self._report_task_run_creation_failure(configuration, task_run_request, exc)
  File "/usr/local/lib/python3.10/site-packages/prefect_aws/workers/ecs_worker.py", line 819, in _report_task_run_creation_failure
    raise RuntimeError(
RuntimeError: Failed to run ECS task, cluster 'default' not found. Confirm that the cluster is configured in your region.

What am I missing? It seems like the task is trying to deploy into the default cluster which doesn’t exist although I have specified where I’d like to deploy the task into.

Hi! i'm trying to log the duration of each flow/task without adding too much code. I see the duration shows up in the UI so it's calculated somewhere, any easy/hacky way to access the start/end times of each task and flow?

Hi folks -- I think I know the answer to this, but is it currently unsupported to run Prefect Server behind a reverse proxy with a prefixed path -- e.g. https://example.com/prefect ? I see a number of bugs and some proposed MRs, but looks like everything has been closed as "won't fix" / "too complicated" and, indeed, I seem unable to get this working locally.

hey all! Is there an easy way to prevent one deployment from having multiple runs? So if it tries to run and it is already running, it will not start the second one?

Hey folks, we've been running into an issue where our Prefect Worker memory usage is constantly increasing until it get to the point where we run out of memory and the service restarts, has anyone experienced this issue or know what is causing this? To give a bit of background, we run our Prefect Worker in a self hosted Kubernetes cluster and it was installed via Helm. Originally we used the default memory limit of 1GB but have since increased to 4GB. The worker is on version 2.11.2

CONCURRENCY GAPS: i have 1000 tasks. concurrency tag set to 20. it runs 20 concurrently then there is this 20-30 second gap before it does another 20 tasks. how do i prevent this time delay “gap” and make it just keep running 20 constantly until done. where is this delay coming from? hard to see in flow screenshot but the little green blobs are the intermittent 20 tasks being completed.

Hi all, I have issues with deleting a flow from the Orion. Does anybody know the fix?

Hello, since the flow has to be unique within a workspace, I would like to alert the developer during PR if they’re trying to use a name that already exists. Are there any apis to validate this using the prefect. yaml . I don’t want a surprise when i run prefect deploy command Any suggestions ?

Hi folks, did someone have same issue or can help me to solve it? I have a flow that run 2 dbt projects one after another as subflows in one flow. The storage of flows and dbt projects is GitHub. Prefect Agent is running on Azure VM. First dbt flow runs always successfully while second flow sometimes has this error and I can’t find the reason why. Any ideas? Thanks a lot!

Finished in state Failed('Flow run encountered an exception. shutil.Error: [(\'C:\\\\Users\\\\user_x\\\\AppData\\\\Local\\\\Temp\\\\tmpu8dj3ri0prefect\\\\.git\\\\objects\\\\pack\\\\pack-dcb61de38801bc82e121d9ddd04df4adace353d8.idx\', \'C:\\\\Users\\\\user_x\\\\AppData\\\\Local\\\\Temp\\\\tmpj3paxn4kprefect\\\\.git\\\\objects\\\\pack\\\\pack-dcb61de38801bc82e121d9ddd04df4adace353d8.idx\', "[Errno 13] Permission denied: \'C:\\\\\\\\Users\\\\\\\\user_x\\\\\\\\AppData\\\\\\\\Local\\\\\\\\Temp\\\\\\\\tmpj3paxn4kprefect\\\\\\\\.git\\\\\\\\objects\\\\\\\\pack\\\\\\\\pack-dcb61de38801bc82e121d9ddd04df4adace353d8.idx\'"), (\'C:\\\\Users\\\\user_x\\\\AppData\\\\Local\\\\Temp\\\\tmpu8dj3ri0prefect\\\\.git\\\\objects\\\\pack\\\\pack-dcb61de38801bc82e121d9ddd04df4adace353d8.pack\', \'C:\\\\Users\\\\user_x\\\\AppData\\\\Local\\\\Temp\\\\tmpj3paxn4kprefect\\\\.git\\\\objects\\\\pack\\\\pack-dcb61de38801bc82e121d9ddd04df4adace353d8.pack\', "[Errno 13] Permission denied: \'C:\\\\\\\\Users\\\\\\\\user_x\\\\\\\\AppData\\\\\\\\Local\\\\\\\\Temp\\\\\\\\tmpj3paxn4kprefect\\\\\\\\.git\\\\\\\\objects\\\\\\\\pack\\\\\\\\pack-dcb61de38801bc82e121d9ddd04df4adace353d8.pack\'")]\n')

Hey Community, I am trying to use Marvin in a project, but I can see that prefect 2.0 does not like

pydantic > 2.0

and marvin requires

>= 2.0

Any chance this will be getting resolved any time soon?

This message was deleted.

I am deploying my Prefect flows in Docker. I have the deployment deploying correctly, and when I create a job it is being picked up by a worker and running in Docker correctly. Two questions • How do I change the docker networking to "host" for the running container (I want to be able to access a service locally on the same host)? • How do I have the container delete itself after a run?

Hi! I am using the work pool Azure Container Instance:push. For specific deployments which uses this work-pool, how can I override the docker image url?

Hello! I am currently running into issues when using @tasks in a flow, we are iterating through a lot of dates and pass very large pandas dataframes to a task for each date, what I notice is that when you do this after about the 10th day our flow gets OOMKilled, if we don't use tasks it doesn't seem to have an issue. The problem being our upsert method is currently a task. My assumption is that the flow is keeping track of the parameters sent to each task and effectively running out of memory by trying to keep each of our dataframes in memory. Is there a way to avoid this or any information on this kind of thing? 🙂

Hello, Is there a way to make the flow end in a failed state if any logger.error was printed or if any exception was raised?

Is there a good example of how to do subflows on your local account & on a local server? The documentation is unclear about what you would do if you had 1 flow in Python File 1 and 1 in Python File 2 and you wanted to schedule them as subflows together in Python File 3.

Prefect 2.0 migration newbe question ... I created a Organization, with one workspace (added billing and paid with credit card for new users) but when I invite other team members it asks them that there is no organization and asks them to create a space

I am new here and working on understanding migration from Prefect 1 to 2. Is Prefect 1 officially deprecated or just not receiving support from the team anymore?

I'm trying to figure out how to use the prefect cli to get secrets

marvin told me this

but .. he lies..

any thoughts?