Hi! I try to implement custom retrying policy in Prefect Orion (2.0b2). But, as i see at this moment, policies in Orion are hardcoded (CoreFlowPolicy/CoreTaskPolicy, MinimalFlowPolicy/MinimalTaskPolicy) and server API handlers for setting states do not have parameters in requests to receive data about custom policies (task policy and flow policy not a parts of requests, but just a dependency injection parameters initialised with

provide_task_policy

/

provide_flow_policy

). So my question is about orchestration settings that will be released in the future. Is custom orchestration policies will be parameters of API requests to set state of task/flow? Or orchestration concepts and APIs will be seriously changed in the near future? How orchestration policies will be serialized/deserialized between server and agent’s engines?

Hi everyone 👋 ! I'm using prefect since a few months now but I keep encountering the same issue with my Airbyte tasks: From some reasons, my Airbyte connection failed at some point and throw a

Failed

status but, I don't know why, prefect considers the task as successful (cf. my screenshot in attachment). It's a blocking point from my point of view as it prevents me from relying on prefect and therefore forces me to double-check each time with Airbyte. Maybe this issue comes from my code but I don't see why ?

sync_airbyte_connection = AirbyteConnectionTask(
    max_retries=3, retry_delay=timedelta(seconds=10), timeout=timedelta(minutes=30),
)

with Flow("my flow", run_config=UniversalRun()) as flow:
    airbyte_sync = sync_airbyte_connection(
        <connection_infos>,
    )
    [...]

flow.set_reference_tasks([
   airbyte_sync
])

Hi, if anyone is experiencing issues with EKS and/or Kubernetes on dask using prefect, I found that the latest release (14h ago) of kubernetes-asyncio (22.6) has been breaking all my flows that worked yesterday (it fails to pickup serviceaccount for rbac, sends all requests as 'system/anonymous'). Pinning version 'kubernetes-asyncio<22.6' in my requirements fixed it for me. Issue: https://github.com/PrefectHQ/prefect/issues/5573

Hey. In test I would like to assert that create_flow_run was called with expected parameters. Is it possible somehow to get task result?

We've started seeing an error when accessing secrets from Prefect Cloud. Any ideas how I can troubleshoot this?

prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

We've laughed, we've cried, but most of all we've 🚀 launched! Come join @Jeremiah, @Chris White and myself as we recap Launch Week 2022 at 3p Eastern on . You might find yourself hearing a few more exciting announcements 😏 https://prefect-community.slack.com/archives/C036FRC4KMW/p1647621014192929

I'd like to set up a testing framework for testing Prefect flows. The idea is that when someone modifies a Prefect flow in a pull request, an automated DevOps process can launch that flow with certain parameters and somehow test that the flow did what it was supposed to do. I couldn't find anything in the documentation or in Slack. Is there an official testing framework for Prefect flows? If not, what do you all use?

Hi, I am making my first pass at using Github actions to CI. I have reached something I don’t fully understand. On pull_request of a feature/** branch, I run actions that: 1. checkout the branch 2. create a project in prefect cloud 3. register the flow.py with a --label dev when I run the flow manually, i get an error:

Unexpected error while running flow: KeyError('Task slug connect_source-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did you change the flow without re-registering it?\n- Did you register the flow without updating it in your storage location (if applicable)?')

See you all in 5 mins on a Fireside Chat:

Hi, do you guys know if it’s possible to pass

upstream_tasks

with methods that actually requires parameter? The documentation is not really clear regarding how to chain multiple upstream_tasks to create dependencies between them and I was wondering if some of you had some more complex examples. For example, I don’t think I can do this

with Flow(f"{environment_prefix}-test", storage=S3(bucket=storage_location_bucket_name)) as flow:


    dbt_run(organization_id_param, data_processing_start_date_param, data_processing_end_date_param, should_process_last_period, period, period_value,
            upstream_tasks=[pull_snowflake_secret(a,b), pull_repo(b,c)])

Does anyone have a complete working example on how to use PrefectResult? Its simple: Scrape a website for a link, if the link is different than the prefect result, download it and save the new link as the result. Otherwise exit with "success: no work to be done". I don't want this to be a local result because we plan to move these flows off of a windows agent and into docker agent. Thanks in advance!

is there an equivalent of

IMAGE_PULL_SECRETS

for the Orion k8s agent/flow runner? I couldn’t find anything in the docs and the code

Hello, for prefect 2.0 - if I want to add flow run id in all the log statements, do I have to override format of all the loggers defined in default logging config or is there a way to apply this change by overriding only one parent logger?

Hi all! I am new developer and an new to prefect. Currently we are trying to implement prefect for our ETLS at my job. We use vmware and windows sever, we found out the hard way that using prefect with nested virtualization does not work well. So now are moving to prefect cloud, using local agents. Will this work? Is it necessary to run multiple agents for multiple flows or can one agent run multiple flows?

Hi, in prefect 0.14.11, prefect execute flow-run downloads the necessary local dependencies from the repo when using GitLab storage but this behavior seems to have changed in at least 1.0.0…could you help remind me of when this was changed? Also does this mean that we have to build a docker image for each repo we have? :(

does configuring a flow with

KubernetesRun

restrict the flow only to kubernetes agents or can it still be picked up by any agent as long as the labels match? I was under this impression after reading the docs but when I created 1 docker agent and 1 k8s agent with the same set of labels, my kubernetes flow got picked up by the docker agent

Has anyone used Great Expectations, how does it compare to other testing frameworks?

Hi all! I followed the "Running flows in Docker" tutorial for 2.0 (

prefect==2.0b2

, Python 3.9, Linux): • created an S3 bucket and a storage (isn't it too much for a local run?) • started a local API server and configured PREFECT_API_URL • created a queue • started an agent for that queue • created a deployment with DockerFlowRunner and no parameters Then I ran the deployment and the agent output wasn't very informative, see below. In the UI the flow run "dag-sample/caped-antelope" is shown as failed without any logs available. Anyone can explain?

$ prefect agent start 6aa0c2b2-e895-4b8c-aed3-8bef5b2c88ab
Starting agent connected to <http://127.0.0.1:4200/api>...

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


Agent started! Looking for work from queue '6aa0c2b2-e895-4b8c-aed3-8bef5b2c88ab'...
13:29:31.446 | INFO    | prefect.agent - Submitting flow run '8eb8aa0b-82da-4081-9a42-aae1f22b0525'
/home/tarkovskyi/miniconda3/envs/prefect_exp39/lib/python3.9/site-packages/prefect/flow_runners.py:697: UserWarning: `host.docker.internal` could not be automatically resolved to your local ip address. This feature is not supported on Docker Engine v19.3.15, upgrade to v20.10.0+ if you encounter issues.
  warnings.warn(
13:29:32.215 | INFO    | prefect.flow_runner.docker - Flow run 'caped-antelope' has container settings = {'image': 'prefecthq/prefect:2.0b2-python3.9', 'network': None, 'network_mode': 'host', 'command': ['python', '-m', 'prefect.engine', '8eb8aa0b-82da-4081-9a42-aae1f22b0525'], 'environment': {'PREFECT_API_URL': '<http://127.0.0.1:4200/api>'}, 'auto_remove': False, 'labels': {'io.prefect.flow-run-id': '8eb8aa0b-82da-4081-9a42-aae1f22b0525'}, 'extra_hosts': {}, 'name': 'caped-antelope', 'volumes': []}
13:29:33.547 | INFO    | prefect.agent - Completed submission of flow run '8eb8aa0b-82da-4081-9a42-aae1f22b0525'
13:29:33.584 | INFO    | prefect.flow_runner.docker - Flow run container 'caped-antelope' has status 'running'
13:29:44.923 | INFO    | prefect.flow_runner.docker - Flow run container 'caped-antelope' has status 'exited'

I'm struggling to implement the GCP_CREDENTIALS default secret feature in a local Prefect flow. • I've set an environmental variable PREFECT__CONTEXT__SECRETS_GCP_CREDENTIALS pointing to a full local path for the GCP credentials .json downloaded from the GCP project to which I'd like to upload some files. • I've run the following:

from prefect import Flow
from prefect.tasks.gcp.storage import GCSUpload

with Flow("google-cloud-test") as flow:

    GCSUpload(bucket="test-upload", create_bucket=True)(
        data="test-file.csv", credentials="GCP_CREDENTIALS"
    )

flow.run()

and received an error:

AttributeError: 'str' object has no attribute 'keys'

. I've incorporated the feedback from a similar error discussion recently, but can't figure out what the issue is. Any help is much appreciated.

Hi Everyone, we have a prefect server running on Kubernetes, setup was done using the HELM Chart. Everything is running smoothly but occasionally we would get this error

File "/usr/local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 341, in _raise_timeout
    self, url, "Read timed out. (read timeout=%s)" % timeout_value
urllib3.exceptions.ReadTimeoutError: HTTPConnectionPool(host='prefect-apollo.prefect', port=4200): Read timed out. (read timeout=15)

This happens especially when we use the Task StartFlowRun it does not happen very often , but I was wondering if there was a way to force retry or if anyone knows why this would be happening? Thanks

Is there a way to get logs from code running within dask? I know that for "general" use of dask, the logs would have to be pulled from the execution environment such as k8s (as noted here) When I am running code that uses

worker_client

and then submit new tasks to the same cluster (code inside), I see that printout from the first call of the function is logged in prefect and shipped to server, but consecutive calls are not (see log inside). Is there some trick I can use?

Is there an example template of what a properly architected git repo looks like for a Prefect project? I am having trouble conceptualizing the development workflow. I am imagining a "production" branch that at any given moment contains a portfolio of all the relevant flows for a given project. Changes (i.e. merges) to the production branch would trigger incremental registrations of the modified flows driven by some sort of backend "build" job triggered by the successful merge. Other things I am considering would be a way to have proposed changes to production flows go through automated testing before being merged to production and ideally a way for developers to test their flows in development without sullying the production environment. Bonus points for being able to code the tasks and flows in such a way so that they do not require editing between environments. is there a recognized reference architecture for such a build strategy?

Hi, I want to de-register some flows programatically based on certain conditions. Is it possible to: 1. Get a list of all registered flows in python (i.e. the Python equivalent of

prefect get flows

), and 2. De-register some of these flows through python.

Hi - is there an explanation of what exactly Workspaces in Prefect Cloud 2.0 do/are for? The docs I could find say how to create one but not really why you’d want to.

hey everyone, looking to change our ETL flow from celery alone to prefect, and i came across this discussion on github: https://github.com/PrefectHQ/prefect/issues/1689 we are looking to have the easiest transformation into prefect and would like to have minimum changes, but i could not find a way to use a CeleryExecutor like Dask has. Am i missing something? is it supported or should i implement it on my own? Thank you 🙂

Hi! What is the equivalent of using a json prefect secret in the cloud, if I want to do this in the local secret (.prefect/config.toml)?

Hi! How do I mount AWS credentials to

KubernetesFlowRunner

?

Lots of good fixes in 2.0b2

https://orion-docs.prefect.io/concepts/tasks/#configuring-concurrency-limits On this page it says to use prefect concurrency_limit; however, the actual command is concurrency-limit