prefect-community #prefect-community

prefect-community

Michael Dymshits

12/20/2022, 12:51 PM

When I import function for

pytest

which has perfect

decorator

, I got the following error (screenshot bellow). If I comment #flow decorator, pytest runs without errors.

Michael Dymshits

12/20/2022, 12:52 PM

hkazmi

12/20/2022, 1:22 PM

Hello everyone, I am not sure if this is correct place to ask this question, but today suddenly all of our prefect flows started failing and raise following error

(401) Reason: Unauthorized HTTP response headers: HTTPHeaderDict({'Audit-Id': '', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'Date': 'Tue, 20 Dec 2022 13:00:03 GMT', 'Content-Length': '129'}) HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"Unauthorized","reason":"Unauthorized","code":401}

I have checked all keys and non of them is expired What could be other possible reason for this error ?

👀 1

✅ 1

Harry Hause

12/20/2022, 3:05 PM

Hey @Anna Geller , I’ve been following your repo here to move our current deployment from Prefect 1.0 on ECS to 2.0 running on K8s with CD. Thanks for taking the time to put that together. It’s been really helpful so far. One question I have is related to expanding the CD to multiple flows. We currently have multiple flows registered manually on Prefect 1.0 and if we could automate the deployment for all of them that would be great. Ideally we would like to keep all of our flows in 1 repo. Would we just add more steps to the github workflows for each flow?

Marcelo Santoro

12/20/2022, 3:13 PM

Hey Prefect community .... Does anyone knows if .... is there any way to rename the k8s prefect jobs ? because is really hard to identify what flow is running in each job 😕

👀 1

✅ 1

Vadym Dytyniak

12/20/2022, 3:14 PM

Hi. My question is related to prefect-dask collection. Depends on the image I use for my DaskCluster Prefect submits tasks to DaskTaskRunner with different behavior. My task does manipilations with dask.dataframe and in one case dask transformation is distributed on the same cluster, but in other case task is running on one worker and other just waiting. Does anyone has an idea what should be different to see such different behaviour? Thanks.

Missi Sogbohossou

12/20/2022, 3:49 PM

Good morning! I'm using Prefect 1.0. The logs have not been loading to the UI for a couple of days and I was wondering if there has been any changes.

Missi Sogbohossou

12/20/2022, 3:49 PM

👀 1

✅ 1

Madison Schott

12/20/2022, 5:09 PM

Hi all, getting this error: I created an exact copy of my current flow in production but gave it a new name and wrote it in a new Python file. I removed some of the syncs and other commands but not any of the fundamentals so not sure what could be causing this,

/usr/local/lib/python3.9/site-packages/prefect/core/flow.py:1726: UserWarning: No result handler was specified on your Flow. Cloud features such as input caching and resuming task runs from failure may not work properly.

Edmund Tian

12/20/2022, 5:29 PM

Hi, I’m getting the following error during my Flow Run.

OperationalError:
(sqlite3.OperationalError) no such table: flow_run_alert_policy
[SQL: ALTER TABLE flow_run_alert_policy RENAME TO flow_run_notification_policy]
(Background on this error at: <https://sqlalche.me/e/14/e3q8>)

I’m using the

@flow

and

@task

decorators for my Python functions. I’m not using Prefect Cloud. I’m on the following prefect version

prefect==2.3.2

. It’s hard for me to repo because this occurs during my chron job, and it only occurs for a small subset of the Flow Runs. I haven’t been able to repo locally. Anyone know what’s going on? A quick Google brought me to this, but this doesn’t seem related as I am not deploying anything and I’m not using

orion

Patrick Tan

12/20/2022, 6:57 PM

Hi, I have a scenerio where agent not picking up flow run when work queue concurrency is set to a number( I use 5 and 25). However, if I change to unlimited it is picking up flow run and execute. I do not have any flow run that is in 'running' state

Patrick Tan

12/20/2022, 7:09 PM

nvm, found the issue, in addition of 'running', 'pending' flow run is also counted for concurrency

Patrick Tan

12/20/2022, 7:13 PM

All, I want to delete 200+ flow run that are in pending state, I can filter them in UI. However, I am not able to 'select all' and delete all at once. I have to select one check box at a time. Any short cut ?

👀 1

Amruth VVKP

12/20/2022, 8:08 PM

Need help deploying Prefect Orion and Prefect Agent to ECS Fargate Clusters - Prefect 2.7.3 =========================================================================== Hi, I am trying to deploy Prefect Orion and Prefect Agent to ECS Fargate clusters. I've had a successful deployment with Docker-Compose but somehow I've been running around in circles deploying it to Fargate. Here's what I am trying to do - 1. Prefect Orion server - running on a Fargate task 1 with Public IP 2. Prefect Agent 1 - running on Fargate task 2, looks up to the deployed Orion server 3. Prefect Agent 2 - running on Fargate task 3, looks up to the deployed Orion server Before anything moves further, I plan on upgrading future tasks to run as ECS tasks (still need to play around with that, but it's something for later) What I have done so far is - create an ECR image built from custom flows and tasks that we have put in, create - 1. ECS Cluster 2. ECS Service and task for Prefect Orion that uses latest Prefect image from Docker-Hub 3. ECS Service and task for Agent-1 that uses custom image from ECR repo built over Prefect's latest docker image 4. ECS Service and task for Agent-2 that uses custom image from ECR repo built over Prefect's latest docker image I've put in a AWS CloudMap to generate a internal network that allows Prefect's agents to look up for Prefect Orion server using internally resolved DNS. This is running into few issues - 1. Prefect Orion starts but throws an error - Can't connect to Orion API at <CloudMap resolved DNS>/api or <CloudMap resolved DNS>:4200/api or 0.0.0.0:4200/api. The task definition does contain Prefect API listed as an environment variable and PostgreSQL DB fetched from AWS SecretsManager. The web UI is running up but can't seem to connect to the DB or show anything unders /setings. (Attached sreenshot) 2. Prefect agents on both the tasks are not starting up and fail with this error - Resourceinitializationerror: unable to pull secrets or registry auth: execution resource retrieval failed: unable to retrieve secret from asm: service call has been retried 5 time(s): failed to fetch secret arn:aws:secretsmanager:_**_** from secrets manager: RequestCanceled: request context canceled caused by: context deadline exceeded. The same secret passed to Prefect Orion Fargate task seem to work just fine.

Santhosh Solomon (Fluffy)

12/20/2022, 9:35 PM

Hi all, pardon me for the naïve question. The concurrency limit for the queue in prefect 2 is for setting the concurrency of the tasks submitted in the flows or number of flow run executions per deployment?

Adam Roderick

12/20/2022, 10:41 PM

Is this a good place to ask questions about the prefect-snowflake task? It uses

.fetchall()

which isn't usable for large datasets. Any other attempts we've done to use similar logic but using a cursor fails. What is it about the prefect execution environment that prevents us from using snowflake cursors?

Samuel Hinton

12/21/2022, 2:04 AM

Hey team! We’re looking at upgrading from Prefect1 to 2 in the near future. Currently our prefect services are deployed in house (using terraform and what not), but this allows us to quickly spin up and down new testing environments which use prefect 1. However, I notice in prefect 2, there are the new concepts of storage blocks, and if we want to continue to use s3 as a host for our flows, it seems like we have to login to a newly created environment and manually create the s3 storage block. Is there a way (through the API, env variables, other) we can set up an s3 bucket to allow us to keep the entire setup as infrastructure-as-code?

Niwde Nus

12/21/2022, 2:39 AM

Hi all, what does this mean? "Local storage is not supported for Docker" Does Prefect store the docker images on the specified remote storage block? Why cant it store it on a local drive?

✅ 1

Gows Shaik

12/21/2022, 5:38 AM

Hi eveyone,

Gows Shaik

12/21/2022, 5:40 AM

Hello everyone,Iam not able to connect to prefect cloud via my PC although i have all access.Its showing SSL error wheen i have connected to my organization VPN.Its connecting fine when iam not connected VPN.Can anyone suggest how to access prefect cloud while connecting to VPN.

Malavika S Menon

12/21/2022, 6:06 AM

Hi, we had configured Slack notifications to be triggered on failure and success. However we're noticing failure messages arent getting reported. What could be the potential reason for this? We're running a self hosted prefect server

Deepanshu Aggarwal

12/21/2022, 6:35 AM

hi ! some of my flows have started crashing with

Flow run infrastructure exited with non-zero status code -1.

no extra logs nothing, it also leaves the tasks in running state. any idea why this happens ?

12/21/2022, 7:41 AM

Sending SMS on task failure - Prefect 2.0 - Plivo I use Plivo as SMS service, not Twilio in Prefect 1.0 I used

'state_handlers': [o.p_state_handler]

which sent me an SMS alert on task failure. how can i achieve this kind of behavior in Prefect 2.0? thanks

12/21/2022, 7:51 AM

Prefect 2 task timeout_seconds looking at https://docs.prefect.io/concepts/tasks/#timeouts it looks like this should be allowed:

@task(timeout_seconds=1)

but I get an error:

TypeError: task() got an unexpected keyword argument 'timeout_seconds'

What is the proper way to set a timeout time for tasks in Prefect 2 ?

✅ 1

Kelvin DeCosta

12/21/2022, 9:31 AM

Quick question: If a

task

is configured with

cache_result_in_memory=False

, can its result only be consumed via

state.result()

future.result()

? Calling such tasks directly, i.e..

result = my_task()

, seems to raise the

MissingResult

error.

Mark Nuttall-Smith

12/21/2022, 9:48 AM

Hi, I'm trying to understand how remote-storage should be used to distribute flows to a heterogeneous execution environment. I've drawn a couple of diagrams to clarify my understanding. The first diagram shows the creation of a deployment in the orion server via the cli. Config goes to orion, code goes to the s3 remote storage bucket. The agent subscribes to the work queue. So far so good. The second diagram shows the prefect agent receiving an run event from the queue and spawning a

myapp

docker container for execution.

Aleksandr Liadov

12/21/2022, 11:15 AM

Hello, Anybody had already any trouble with the Dask order of execution? (Local dask execution) My flow is executed right with ConcurrentTaskRunner, but DaskTaskRunner executes tasks, which should be executed only when their dependencies are executed. First image -concurrent graph (correct), second dask one (incorrect) I have created a relative issue

Jaime Raldua Veuthey

12/21/2022, 11:32 AM

Hi, we are considering using Airtable for a DB. We would need users who can access through UI (read only access) and also programmatic access and because Airtable is so user-friendly it is on our list to be considered. Has anyone experience accessing programmatically Airtable via API? I heard the API is not very good and only supports HTTP. Have you worked with any other DB that would fulfil these requirements? user-friendly UI for read-only-user and accessible securely via API Also any suggested link about how to orchestrate the input/output from Airtable with Prefect? Thanks in advance!

✅ 1

Alexandru Anghel

12/21/2022, 1:58 PM

Hello guys, I get the following error when trying to run a flow using the DaskTaskRunner using KubeCluster:

Crash detected! Execution was interrupted by an unexpected exception: AttributeError: 'KubeCluster' object has no attribute '__qualname__'

For more context, i'll add some code in the thread. Thanks!

✅ 1

Michał

12/21/2022, 2:24 PM

Hey, guys I have problem with agent after succesffully runned flow (all 'tasks" run and finish but seting state fails:

For more information check: <https://httpstatuses.com/422>

15:18:47.351 | INFO | prefect.agent - Completed submission of flow run '5adc8699-d9a5-4042-85d0-0b072a4873d0' <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 15:18:48.970 | INFO | Flow run 'fantastic-tamarin' - Hello from demo-flow! 15:18:48.995 | ERROR | Flow run 'fantastic-tamarin' - Crash detected! Execution was interrupted by an unexpected exception: httpx.HTTPStatusError: Client error '422 Unprocessable Entity' for url 'http://prefect-server:4200/api/flow_runs/5adc8699-d9a5-4042-85d0-0b072a4873d0/set_state'

Title

Michał

12/21/2022, 2:24 PM

Hey, guys I have problem with agent after succesffully runned flow (all 'tasks" run and finish but seting state fails:

For more information check: <https://httpstatuses.com/422>

Khuyen Tran

12/21/2022, 5:17 PM

Hi @Michał, do you mind shortening your error message and post the full error in the comment to keep the channel clean? Can you also share an example of the code that causes the error?

Michał

12/21/2022, 5:18 PM

@Khuyen Tran Yeah, ofcourse For more information check: https://httpstatuses.com/422 15:18:49.058 | ERROR | prefect.engine - Engine execution of flow run '5adc8699-d9a5-4042-85d0-0b072a4873d0' exited with unexpected exception Traceback (most recent call last): File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 1898, in <module> enter_flow_run_engine_from_subprocess(flow_run_id) File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 188, in enter_flow_run_engine_from_subprocess return anyio.run(retrieve_flow_then_begin_flow_run, flow_run_id) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/anyio/_core/_eventloop.py", line 70, in run return asynclib.run(func, *args, **backend_options) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 292, in run return native_run(wrapper(), debug=debug) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/runners.py", line 190, in run return runner.run(main) ^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/runners.py", line 118, in run return self._loop.run_until_complete(task) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete return future.result() ^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper return await func(*args) ^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/client/utilities.py", line 47, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 306, in retrieve_flow_then_begin_flow_run return await begin_flow_run( ^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/contextlib.py", line 222, in aexit await self.gen.athrow(typ, value, traceback) File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 1594, in report_flow_run_crashes raise exc from None File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 1574, in report_flow_run_crashes yield File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 372, in begin_flow_run terminal_or_paused_state = await orchestrate_flow_run( ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 697, in orchestrate_flow_run state = await propose_state( ^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/engine.py", line 1741, in propose_state response = await client.set_flow_run_state( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/client/orion.py", line 1549, in set_flow_run_state response = await self._client.post( ^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/httpx/_client.py", line 1848, in post return await self.request( ^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/httpx/_client.py", line 1533, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/prefect/client/base.py", line 251, in send response.raise_for_status() File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/httpx/_models.py", line 745, in raise_for_status raise HTTPStatusError(message, request=request, response=self) httpx.HTTPStatusError: Client error '422 Unprocessable Entity' for url 'http://prefect-server:4200/api/flow_runs/5adc8699-d9a5-4042-85d0-0b072a4873d0/set_state' For more information check: https://httpstatuses.com/422 15:18:49.339 | ERROR | prefect.infrastructure.process - Process 'fantastic-tamarin' exited with status code:

Ok I guess i found problem. When i check /api/

curl -X 'GET' \
  '<http://prefect-server:14200/api/admin/version>' \
  -H 'accept: application/json'

I found that even I use tag 2,7.3 i get version 2.4.5 in this endpoint when I downgraded my local prefect version from agent from 2.7.3 to 2.4.5 flows works and I don't get this 422 error. But when I got back to version 2.7.3 error is back.

Tibs

01/09/2023, 3:11 PM

I get this error after upgrading from 2.7.0 to 2.7.6..

Taylor Curran

01/09/2023, 5:39 PM

Updating to prefect-2.7.7 fixed this for me @Tibs

✅ 1

View count: 4