Hello! I have a v1 flow which ingests data via SFTP. I have a task which lists the files to be downloaded, and then a mapped task which downloads each file. The flow completed successfully, but seems to have not downloaded a few files. There were 3,202 files, but only 3,197 seem to have been downloaded. So we're missing 5. There are no errors associated with this; rather, it appears that only 3,197 mapped runs were ever scheduled, even though it says "Expected Runs: 3,202". When I mouse over the "3,202", a popup says:

The number of mapped children expected to run.

Note that the number of active mapped runs may be less than this if some have not yet entered a

Pending

state.

This seems like it would make sense while the flow was still running, but not once it has completed. 🤔

I have Prefect orchestrate the soda scan, I will always get this error if one of my checks fails

Hello, I've been playing around with trying to get the flow logs we see in the UI saved onto disk or on S3. Currently I have my set up like this :

@flow
def api_flow(url):
    # logger = get_run_logger()
    logger = logging.getLogger()

    #set lower level
    logger.setLevel(<http://logging.INFO|logging.INFO>)

    #create file handler
    filehdlr = logging.FileHandler('filehandler.txt',mode='w')
    # set level
    filehdlr.setLevel(<http://logging.INFO|logging.INFO>)

    # add handler to logger
    logger.addHandler(filehdlr)
    fact_json = call_api(url)
    <http://logger.info|logger.info>("TEST")
    return fact_json

But it only adds local outputs into the .txt file (filehandler.txt shown below):

file handler is created to handle log message in the file
we can create many handlers for the logger
Created task run 'call_api-0' for task 'call_api'
Executing 'call_api-0' immediately...
Finished in state Completed()
TEST
Finished in state Completed()

Is there a way I can get the exact flow logs (including the timestamps) saved somewhere (either onprem or on S3) like the output shown below from prefect?

I have a service that receives HTTP requests (Kafka topics in the future) that triggers some tasks in celery, with all bells and whistles expected from a tasking management system (canceling, monitoring, cache, retry). This above description strikes me as a perfectly acceptable job for Prefect, am I correct? I mean, could I replace all that "engineering" above with a Prefect instance and only be concerned with writing the tasks themselves? Or would I be using Prefect in an "incorrect" way or in an "unforeseen" way? Besides that, does the community have any resources or examples on how to achieve that? I'm mostly looking for examples of how to interact with the flows/deployments with API as a service and on best practices and coordination planes.

Docker + Prefect: Does Prefect handle removal of a docker container after a flow is ended (successfully or not)? I am worried because for each retry of a flow, I see a new container created, with increasing serial number at the end.

Hi @Prefect, in the recent weeks, Prefect has release an incredible number of features - And ALL of them were on my wishlist when we migrated to prefect earlier this year. Is there a published mid/long-term roadmap for the product ? curious and eager to see what’s coming up next A big thanks to the Prefect team !!!

Curious where this default storage path lives https://docs.prefect.io/concepts/results/#result-storage-location

PREFECT_LOCAL_STORAGE_PATH

setting (defaults to

~/.prefect/storage

).

I don’t see this on the VM we’re running our prefect agents and docker containers on that run the flows. Since we don’t have the

PREFECT_LOCAL_STORAGE_PATH

env var set, is the

~/.prefect/storage

on prefect servers?

The work pools doc says the pools are created automatically when configured on an agent. What I’m seeing is when I specify a new work pool name in the agent args, it crashes with

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url 'http://<host>/api/work_pools/data-qa/get_scheduled_flow_runs'"

, i.e. fails on a missing work queue. What am I doing wrong? both server and agent are 2.9.0

Hi all! What are the best practices when it comes to needing to run large sql statements with parameters in a task/flow? In airflow we can pass an sql file with jinja templating. Can something similar be done with prefect?

Hi all, How do I tag a flow which is being deployed? How do I also tag a sub flow without tagging the tasks under the sub flow? Using

with prefect.tags

effects both the sub flow and the tasks under the sub flows.

Hi Pteam, love the new artifacts! Is there a way to link artifacts from the flow-run view? Such that I can go from a task-run to the artifact?

HI all, 🐛 🐛 🐛 🐛 🐛 I think ive found a bug. When querying for

task concurrency tag

by tag name if the tag doesnt exist , wrong exception is being raised . Exception being raised :

prefect.exceptions.PrefectHTTPStatusError

Exception that should have been raised :

prefect.exceptions.ObjectNotFound

Prefect 2.8.4: Is it possible to "group" deployments (even just in the UI)? For example, I have numerous deployments for a certain project, and I would like to group them under, say, an expansion panel, so that my deployments page is less cluttered. Is this possible as of now?

hello prefect team, hello community, some flow parameter types are displayed with additional interaction elements on the flow run submission page (arrow up/down for int, dropdown menu for Literal, calendar for datetime, etc.), which is extremely nice. Is there any documentation available that lists the datatypes that have such a special UI representation?

Hello all! I have a deployment that runs a main flow that deploys additional deployments. I was able to have the main deployment run after running into the issue here but my children deployments are failing. Does anyone have any recommendations on how to fix this? Details in thread.

Hey all, I am working on a deployment using github, docker, and ECS. It was recommended to me since you can only have 1 infra block to use it for ECS and do a custom docker container. I am still not super familiar with docker and I am having a hard time figuring out how to do a deploy without using the docker block, as I need to reserve the -ib for the ECS task. Hope this makes sense

Hello all, I’m having some difficulty in having the agent connect to the server when running deployments with the docker infrastructure block , but not when running flows directly. Some context: • I’m using docker compose to spin up a

prefect-server

and

prefect-agent

making sure they are in the same named network. • The agent has the host’s docker socket mounted in volumes so it can spin up parallel containers with the Docker infrastructure block, not via Docker-in-Docker. • Created a custom flow runner image based on the official docker image with flows installed in

/opt/prefect/flows

and all the python dependencies installed. • I have created a Docker infrastructure block from the host machine to have the agent spin up the flow runner containers. After I have registered both the Docker block and a deployment, I have tried to do a quick run of the deployment through the UI, but the flow run crashes. The agent log shows (relevant bits):

12:59:26.941 | INFO    | prefect.agent - Submitting flow run '977526d8-f473-4713-b454-a792df88af5f'
12:59:27.023 | INFO    | prefect.infrastructure.docker-container - Creating Docker container 'meteoric-bird'...
12:59:27.160 | INFO    | prefect.infrastructure.docker-container - Docker container 'meteoric-bird' has status 'created'
12:59:27.450 | INFO    | prefect.agent - Completed submission of flow run '977526d8-f473-4713-b454-a792df88af5f'
12:59:27.453 | INFO    | prefect.infrastructure.docker-container - Docker container 'meteoric-bird' has status 'running'
17:59:28.323 | DEBUG   | prefect.profiles - Using profile 'default'
<frozen runpy>:128: 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
17:59:28.335 | DEBUG   | prefect.client - Connecting to API at <http://prefect-server:4200/api/>
17:59:29.873 | ERROR   | prefect.engine - Engine execution of flow run '977526d8-f473-4713-b454-a792df88af5f' exited with unexpected exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/anyio/_core/_sockets.py", line 186, in connect_tcp
    addr_obj = ip_address(remote_host)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/ipaddress.py", line 54, in ip_address
    raise ValueError(f'{address!r} does not appear to be an IPv4 or IPv6 address')
ValueError: 'prefect-server' does not appear to be an IPv4 or IPv6 address

Some things I’ve tried or confirmed: • The

prefect-agent

and

prefect-server

containers are on the same network. Confirmed by

docker network inspect

. • The

prefect-agent

can see the

prefect-server

container and resolve the name when I use both ping (

ping prefect-server

) and hitting the API (

http GET prefect-server:4200/api/health

using httpie). • I can run a python flow file like this one directly in the

prefect-agent

and it runs fine, communicating the results to the

prefect-server

. • The agent is able to spin up the flow runner container, right before it crashes. I don’t understand why the agent seems to be able to resolve the

prefect-server

hostname when running flows locally, but fails to do so when running the same flow through a deployment. Any suggestions on what may be happening, or what I can do to further debug this issue? Many thanks in advance for your help!

Can Prefect use local docker images? Can Prefect use private docker images?

Hi I'm trying to delete a block with a service account on it in prefect cloud and am getting back

This is somewhat case-specific... I have a piece of code (vpn login) that fails with Prefect running docker, but runs just fine in the same docker image locally. I've set

privileged

to

true

. I wonder how Prefect works with docker under the hood. I don't see why the same code & docker image gives different behavior with/without Prefect.

Hi, I am running a deployment multiple times from a master flow with different parameters.. Today, 2 of the child runs failed with error

Submission failed. IndexError: list index out of range

, does anyone know what may be the cause of this? prefect version 2.8.2 flows run using ECSTask

Hi, I am honestly not really happy with the direction Prefect 2.0 is going... Prefect 1.0 was decent, but the current setup really confuses me... I still did not figure out what the best way would be to schedule existing Kubernetes Cronjobs? I know there is prefect-kubernetes, but according to the Medium post you need an external storage? Why not just the Docker container? I mean all files are baked in there, including persistent volumes and everything... Hence my question: What is the best way to schedule existing Kubernetes Jobs/Cronjobs using Prefect, with the background info that the containers use persistent volumes and env vars? example:

# apiVersion: v1
# kind: PersistentVolumeClaim
# metadata:
#   name: composer-bigtechmomentum-v1-pvc
# spec:
#   accessModes:
#     - ReadWriteOnce
#   resources:
#     requests:
#       storage: 1Mi
# ---
apiVersion: batch/v1
kind: CronJob
metadata:
  name: composer-bigtechmomentum-v1
spec:
  schedule: "42 3 * * *"
  suspend: false
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: composer-bigtechmomentum-v1
            image: guestros/xxxxxxxxxx:latest
            imagePullPolicy: Always
          #   volumeMounts:
          #     - mountPath: /app/persistent/
          #       name: composer-bigtechmomentum-v1-pv

Hello 👋 Is there a way to retry

CRASHED

tasks or to avoid setting them as

CRASHED

? This issue was resolved and we can have Dask to retry

RUNNING

tasks now. But since the prefect

2.8.7

version, prefect detects that the Dask worker will be shut down and sets the task as

CRASHED

. When Dask resend the failing task, we end up with

Task run '...' already finished

.

Hi! We are using Prefect v2 and are having issues regarding setting up the

PREFECT_API_URL

variable. Maybe there's a knowledge gap. Can somebody let me know where I would find

ACCOUNT_ID

and

WORKSPACE_ID

? And is there a way to validate the URL beforehand? Thanks.

Hello. I would like to do a health check on my agent. I think it is reasonable to use the following API for that.

<https://api.prefect.cloud/api/accounts/{account_id}/workspaces/{workspace_id}/work_queues/{id}/status>

Is there an SDK available? Or is there a better way to do health checks?

Hi, is there a way to specify task/flow timeout when timeout appears because of infrastructure (for example lost connection with agent)? My flow gets stuck when it uses too much resources on VM, it can run forever 🥲 I'm using prefect 1

Hi, Could I please get some light on Prefect (version 1) Agent issue : -Have created a new API token (with no expiration date) at Prefect Cloud and updated it to register the flow -Got flows registered successfully, however, when trying to run the flow on Prefect Cloud it doesn't run but goes into 'late runs' and producing this error on cloud watch logs :

prefect.exceptions.AuthorizationError: [{'path': ['get_runs_in_queue'], 'message': 'AuthenticationError: Forbidden', 'extensions': {'code': 'UNAUTHENTICATED'}}]

thanks in advance (edited)

Hello! After upgrading my Prefect Server to 2.9.0 I’m seeing a big increase in memory usage for the server. Provisioning 1024-2048 Mb is not enough. While previous version have ran just fine on 512-1024. This in K8s. What are the recommended specs? Or am I seeing some type of memory issue? Thank you!

After upgrade to 2.9.0, our UI which was is run on K8S on separate pod, stopped working:

File "/usr/local/lib/python3.10/site-packages/prefect/server/database/alembic_commands.py", line 53, in alembic_upgrade

alembic.command.upgrade(alembic_config(), revision, sql=dry_run)

File "/usr/local/lib/python3.10/site-packages/alembic/command.py", line 378, in upgrade

script.run_env()

File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 272, in _catch_revision_errors

raise util.CommandError(resolution) from re

alembic.util.exc.CommandError: Can't locate revision identified by '422f8ba9541d'

Application startup failed. Exiting.

Our main API is running outside K8S on another server (VM with dedicated Postgres) with UI turned off, while pod on K8S is running with UI turned on - env is pointing the API outside K8S. Agents on K8S work without any problem after, while UI can't start. Is this an expected behaviour? It's not the first time that new build destroys something on prod in our environment.

Hi, I'd like to send an alert to our Slack block if a task runs longer than expected. I could do it by spinning up a background thread of course. I think the answers no, but just checking - there's no library functions for doing something similar?