The RemoteFileSystem example for MinIO doesn’t work when I try to build deployment with it as storage. I suspect their also needs to be a folder at the end of basepath or it will complain about malformed xml. Also once I add my folder I get past the xml error, but it tells me the specified bucket does not exist. How do I find out what’s going on? Been blocked on this for some hours now. As context the same MinIO bucket and folder worked in Prefect 1.0 . In Prefect 1.0 I used S3(bucket=S3_BUCKET, key=f”scripts/{flow_name}”, client_options=dict(endpoint_url=S3_ENDPOINT_URL, aws_access_key_id=KEY, aws_secret_access_key=SECRET_KEY). And this worked. Any ideas how to get that working in 2.0?

Hi Question: did you run your Python (ETL) app at the same container as prefect is living or you create separate container for this ?

Could someone help me with this please? I want to be able to set SLAs on my flows to kill long running runs - can't seem to work it out from the automations section https://prefect-community.slack.com/archives/CL09KU1K7/p1669375801736019?thread_ts=1668773970.203059&cid=CL09KU1K7

Currently having issues with flows hanging when calling

get_run_context()

from within a flow. MRE in 🧵

Hey all, I'm still new to Prefect but haven't been able resolve a deployment error. I'm able to successfully run my main flow file (daily_run.py), but when I run

deployment build

I get the following errors (in thread). Perhaps they are related, but at this point I'm stumped. Not sure why it runs but cant be deployed. Any help would be appreciated and I'm happy to provide more info as needed. Thanks!

Is there a most correct way to pass mapped results between tasks in a flow? I have a flow that generates a list of items and then repeatedly calls

map

to pass those items/results to follow on tasks. Consistently, if I have a large number of items (50+) then the entire process will lock up during the follow on tasks and fail to terminate cleanly. In the UI the root job is marked as

crashed

but the tasks are always stuck in a running state. I'm running in a Kubernetes environment as well so the stuck jobs have to be manually cleaned up whenever this occurs as they never reach completion. I think this might possibly be related to @Boggdan Barrientos question from yesterday.

Hey prefect - this is a strange one - I got two errors from flows overnight with

State message: Submission failed. IndexError: list index out of range

Unfortunately there are no logs in the agent with any failures either 🤷 This did lead me to see some other errors in the agent 🧵

The

prefect kubernetes manifest agent

command seems to only take in one queue. I'd like the agent to read from multiple queues, is there syntax I can change or is this a bug? using prefect==2.6.7

(scott/CD-562-prefect-poc) prefect kubernetes manifest agent -q test1 -q test2
apiVersion: apps/v1
kind: Deployment
metadata:
  name: prefect-agent
  namespace: default
  labels:
    app: prefect-agent
spec:
  selector:
    matchLabels:
      app: prefect-agent
  replicas: 1
  template:
    metadata:
      labels:
        app: prefect-agent
    spec:
      containers:
      - name: agent
        image: prefecthq/prefect:2.6.7-python3.9
        command: ["prefect", "agent", "start", "-q", "test2"]
        imagePullPolicy: "IfNotPresent"
        env:
          - name: PREFECT_API_URL

prefect 2.7 is it possible to create a bucket on minio using remote file system block. What I am doing at the moment is; when creating a deployment, I create a bucket on mino(from minio UI) first, otherwise I will get an error "bucket doesn't exist"

strange one here - I was trying to investigate the state object just by playing around with it in a dummy flow, like so:

import time

from prefect import flow, get_run_logger
from prefect.deployments import run_deployment


@flow
def run():
    x = run_deployment(name="dro-gallops-fields/default", idempotency_key=str(int(time.time())), timeout=0)

    while x:
        time.sleep(5)

        get_run_logger().info(x.state)
        get_run_logger().info(x.state_name)
        get_run_logger().info(x.state_type)
        get_run_logger().info(x.state.message)
        get_run_logger().info(x.state.result)


if __name__ == "__main__":
    run()

This does everything I expect in Prefect Cloud but the logs on the parent flow "run" show something like this ( forever - it never exits the Scheduled state and updates in line with its actual state ? ) : While the subflow run has the flow makred as

Complete

- is there a different way to ping the state in prefect 2 ?

Hey folks - what happens, if a flow runs and prefect 2.0 cloud gets updated? 🤓 🤓 I'm asking because I very rarely but still get the following error when running flows - and it now was 2 times that it somehow was in the same timerange when a release note was published 😅 But obviously that's highly speculative on my side - is there anything I can provide or investigate the issue?

Crash detected! Execution was interrupted by an unexpected exception: PrefectHTTPStatusError: Server error '500 Internal Server Error' for url '<https://api.prefect.cloud/api/accounts/bd169b15-9cf0-41df-9e46-2233ca3fcfba/workspaces/f507fe51-4c9f-400d-8861-ccfaf33b13e4/task_runs/5d8fc4a8-5ce7-444d-8749-135dce75d9be/set_state>'
Response: {'detail': {'exception_message': 'Internal Server Error'}}
For more information check: <https://httpstatuses.com/500>

EDIT: Ok I found a thread in the best-practices-coordination - channel - there it was confirmed that there was a small service interruption. Therefore this question is obsolete.

hey! is there a way to read the state from flow run object ( returned by run_deployment function) in python?

Hey ! I'm looking at how we are going to deploy our Orion Server on Kubernetes. prefect-helm looks like a really good starting point but we'd like to put an OAuth2 authentication in front of the server for security i.e. making sure only people from our org have access. Is there a recommended way to do this ? Ideally we'd want to deploy flows/blocks from Gitlab CI which means being able to authenticate from terminal.

This message was deleted.

hola! does anyone know if the snowflake task library connector can support can support private keys with passphrase?

Dear Prefect Community, In Prefect Orion, what is the best way to get the "flow_run_id" (dynamically) in the main flow while running that flow? If you can share any sample code, that would be great,

Hi there, I have dozens of workflows on Prefect Orion (v2.6.5) All my workflows are executed as

KubernetesJobs

. I’ve experiencing an issue when I run workflows from UI. 1. I select a Flow -> RUN 2. It schedules a Flow for the next seconds. 3. But the flow keep on pending forever. 4. It actually creates another flow that runs (with no tags, see the images) On the image every squared are actually the same RUN created twice. The original never run, and it creates a new one that runs.

In prefect v1 is there an option to re-run a single task? We had a single task fail, and none of the other tasks depend on it. So other tasks were already in progress and wanted to allow them to keep running, but it would be nice to re-run the failed one.

Any advice on these errors? Seeing this with Flask + Prefect 2.6.8 running in K8's on GCP. Doesn't seem to affect anything, just gives this weird error in the logs:

Traceback (most recent call last):
ERROR | asyncio - Task exception was never retrieved
"future: <Task finished name='Task-36' coro=<<async_generator_athrow without __name__>()> exception=KeyError(139836083032272)>"
File "/usr/local/lib/python3.8/dist-packages/prefect/utilities/asyncutils.py", line 263, in on_shutdown yield
GeneratorExit
During handling of the above exception, another exception occurred:
File "/usr/local/lib/python3.8/dist-packages/prefect/utilities/asyncutils.py", line 267, in on_shutdown
    EVENT_LOOP_GC_REFS.pop(key)
KeyError: 139836083032272"

Hi! Great to see the enhancements of 2.7! I was wondering if you can schedule a cancellation of flow runs? That would then allow a start and stop time to be configured on the scheduling level.

Hi all – I've started using Prefect for some data flows and liking it so far! I have a few questions however as I want to see if there's a better way to do things: In order for a subflow run to be also sent to the queue, I couldn't find a better way other than deploying that subflow and then triggering it via the API with

deployments.run_deployment()

. However, this makes it not possible to manually retry a failed subflow, is there a better way to do this? My use case is mostly the parent flow fetching inputs and then each subflow run would handle one of the inputs (I'm looping over the task that triggers the subflow deployment runs) Thanks!

😞 I'm trying to install prefect library on my new(ish) laptop and this is failing: pip install -U prefect How can I diagnose what's actually wrong here?

Is is possible to manage two separate prefect environments on the same machine? I've been experimenting with profiles, and setting the PREFECT_ORION_DATABASE_CONNECTION_URL to perhaps define a second sqlite db, but no luck. Even in different profiles my prefect commands are aware of deployments from the other environment. Ideally there would be no overlap between prefect environments, different db, etc.

Has the behaviour of

PREFECT_LOGGING_EXTRA_LOGGERS

changed from

2.6.7

to

2.6.9

at all? I was previously using it successfully in combination with

DaskTaskExecutor

and

dask_jobqueue.SLURMCluster

to capture logs from another package. It seems though with an updated to

2.6.8

or

2.6.9

I have lost this ability. I do see the logs being printed to my slurm output files, and I do see it is formatted in the

prefect

style, but I am not seeing these logs being submitted through to my self-hosted Orion server (as presented by the web UI). Any ideas?

Hey! I've a little problem with long running parent flows: There is one parent flow and two sub-flows. The subflows both need between 2 and 30 hours (yes, yes, I know...). Now, quite often, the parent flow enters state "crashed" with State message "Flow run infrastructure exited with non-zero status code -1.". Interestingly, it always happens around runtime of 4 hours. Fortunately, the subflows still continue to run and they also finish. Therefore, it does not really impact the pipeline, because the pipeline simply succeeds - but it's not really nice, that parent flows. Any ideas what could cause the parent to fail? In the logs of the infrastructure (see thread), everything seems fine. (One thing I have to admit: The flows are generating approx. 3 lines of log per second. Is this the problem?). (Second thing to admit: There are a looot of tasks. About 2.000 tasks). prefect 2.0 cloud, v2.7.0 agents. Infrastructure is a kubernetes job. EDIT: I changed to logs and tasks a little bit to have way less logs and tasks. In total there are less than 100 tasks and about 10 logs per minute. But still the same. The parent flows crashed after 4 hours - and the subflows happily continued.

Following my previous query two days ago about the change in behaviour I found in

PREFECT_LOGGING_EXTRA_LOGGERS

no longer reporting logs through to orion. I went reading through

prefect/logging/configuration.py

and see in #7569 that there was some change to the logic. Now there is a test made against

config['incremental']

before the orion log handler is attached to the logger of the extra module. I am not sure exactly what

incrmental

is in this sense, but when I disable this check to force the orion handler to attach itself to my extra module, things work as expected and logs are streamed to orion. It seems that by default that this

config['incremental']

is set to

True

(at least I have not knowingly set it), and the test made against its value is negative when evaluating whether to attach the orion logger. So my question is what is this

incremental

configurable, and how does one set it? Is the check made against it actually intended to be negated? Once I remove the check my logs for extra modules outlined in

PREFECT_LOGGING_EXTRA_LOGGERS

behave the same as pre v2.6.9 / #7569 @Anna Geller @Zanie - I see both your names on the change -- please don't hate me for tagging you directly 🙂

Hi folks, I am facing an issue unable to remove an agent in prefect 1 (cloud) - I tried doing so via graphql and the UI to no avail

Hello everyone, I wanted to ask are there any GitHub repos available for projects with Prefect because I'm currently learning it and I want to see how does a project built with Prefect looks like

Hi 👋 We have several deployments running in production (scheduled with a CRON expression). All of those deployments share a common piece of code coming from a .py file. if I update this .py file, I now need to run the “apply” command for each of the .yaml files. For example:

prefect deployment apply jobs/selfServiceDaily.yaml

Are there any shortcuts to apply many yaml files at once?

Hello everyone!! Hope you're all fine.