Hi all I m currently running Prefect 2 6 5 on Kubernetes run

Question

Hi all I m currently running Prefect 2 6 5 on Kubernetes running all flows in Kubernetes Jobs with flows baked in docker images I seem to run into task run crashes which cause my flow run to crash while I have retrying on these specific tasks The task that reports the crash is a mapped task over 12 inputs The task is executed in a loop with sleeping in between loop iterations It mostly manages to execute a certain amount of iterations before crashing like 30ish Does anyone have an idea on how I could solve this or what exactly is going wrong Stack trace in thread

Mathijs Carlu · Answer

`Crash detected Execution was cancelled by the runtime environment ` `09 50 37 AM` `machine data 7699a41e 398` `DEBUG` `Crash details ` `Traceback most recent call last ` `File usr local lib python3 8 site packages prefect engine py line 1334 in report task run crashes` `yield` `File usr local lib python3 8 site packages prefect engine py line 1070 in begin task run` `connect error = await client api healthcheck ` `File usr local lib python3 8 site packages prefect client orion py line 204 in api healthcheck` `await self client get health ` `File usr local lib python3 8 site packages httpx client py line 1751 in get` `return await self request ` `File usr local lib python3 8 site packages httpx client py line 1527 in request` `return await self send request auth=auth follow redirects=follow redirects ` `File usr local lib python3 8 site packages prefect client base py line 159 in send` `await super send args kwargs ` `File usr local lib python3 8 site packages httpx client py line 1614 in send` `response = await self send handling auth ` `File usr local lib python3 8 site packages httpx client py line 1642 in send handling auth` `response = await self send handling redirects ` `File usr local lib python3 8 site packages httpx client py line 1679 in send handling redirects` `response = await self send single request request ` `File usr local lib python3 8 site packages httpx client py line 1716 in send single request` `response = await transport handle async request request ` `File usr local lib python3 8 site packages httpx transports default py line 353 in handle async request` `resp = await self pool handle async request req ` `File usr local lib python3 8 site packages httpcore async connection pool py line 218 in handle async request` `async with self pool lock ` `File usr local lib python3 8 site packages httpcore synchronization py line 15 in aenter ` `await self lock acquire ` `File usr local lib python3 8 site packages anyio core synchronization py line 130 in acquire` `await event wait ` `File usr local lib python3 8 site packages anyio backends asyncio py line 1842 in wait` `if await self event wait ` `File usr local lib python3 8 asyncio locks py line 309 in wait` `await fut` `asyncio exceptions CancelledError` `09 50 37 AM` `machine data 7699a41e 398` `ERROR` `Crash detected Execution was interrupted by an unexpected exception Traceback most recent call last ` `File usr local lib python3 8 asyncio selector events py line 910 in write` `n = self sock send data ` `BrokenPipeError Errno 32 Broken pipe` `The above exception was the direct cause of the following exception ` `Traceback most recent call last ` `File usr local lib python3 8 site packages httpcore exceptions py line 8 in map exceptions` `yield` `File usr local lib python3 8 site packages httpcore backends asyncio py line 33 in read` `return await self stream receive max bytes=max bytes ` `File usr local lib python3 8 site packages anyio backends asyncio py line 1274 in receive` `raise self protocol exception` `anyio BrokenResourceError` `During handling of the above exception another exception occurred ` `Traceback most recent call last ` `File usr local lib python3 8 site packages httpx transports default py line 60 in map httpcore exceptions` `yield` `File usr local lib python3 8 site packages httpx transports default py line 353 in handle async request` `resp = await self pool handle async request req ` `File usr local lib python3 8 site packages httpcore async connection pool py line 253 in handle async request` `raise exc` `File usr local lib python3 8 site packages httpcore async connection pool py line 237 in handle async request` `response = await connection handle async request request ` `File usr local lib python3 8 site packages httpcore async connection py line 90 in handle async request` `return await self connection handle async request request ` `File usr local lib python3 8 site packages httpcore async http11 py line 105 in handle async request` `raise exc` `File usr local lib python3 8 site packages httpcore async http11 py line 84 in handle async request` ` = await self receive response headers kwargs ` `File usr local lib python3 8 site packages httpcore async http11 py line 148 in receive response headers` `event = await self receive event timeout=timeout ` `File usr local lib python3 8 site packages httpcore async http11 py line 177 in receive event` `data = await self network stream read ` `File usr local lib python3 8 site packages httpcore backends asyncio py line 35 in read` `return b ` `File usr local lib python3 8 contextlib py line 131 in exit ` `self gen throw type value traceback ` `File usr local lib python3 8 site packages httpcore exceptions py line 12 in map exceptions` `raise to exc exc ` `httpcore ReadError` `The above exception was the direct cause of the following exception ` `httpx ReadError`

Sebastian Pabst · Answer

I have the exact same issue I try to retrieve a bunch of blob files from azure via a map in prefect and some of them crash don t fail with a broken Pipe error I also noticed that crashed tasks are not retried while failed ones do tried to add a non existing file in the loop I am running version 2 6 5 too

Faheem Khan · Answer

Got the same issue but didn t see anything crashing prefect 2 6 6

Benny Warlick · Answer

I am getting similar httpx errors and BrokenPipeError and then a crashed flow My VM also died Google Compute Engine that is running the Prefect agent and running flows locally in the VM Not sure if the VM died first and caused the flow to crash or the other way around

Faheem Khan · Answer

did anyone find any solution

Mathijs Carlu · Answer

Nope I ll file a github issue since my question was kind of skipped by the team

Andreas Nigg · Answer

Mathijs do you have an example flow which can easily reproduce this issue I also had this problem once but could not really reproduce it in a minimal example and the prefect team had a hard time to do something as also they were not really able to reproduce as well So if you could add an example flow to reproduce to the github issue this would be awesome I guess Would you mind to link the github issue here I d like to follow the issue as well

Mathijs Carlu · Answer

``` flow persist result=True result storage= azure default def flow input date Union datetime None =None logger = get run logger user = pwd = logger name = logger pwd = machine speed = Martin diSpeed Transline diSpeed Compact diSpeed Ward diSpeed Geopfert diSpeed Straplijn rSpeed Emba diSpeed Magnus diSpeed Mark2 diSpeed Rapidline diSpeed Mark1 diSpeed Copar Speed timezone = pytz timezone Europe Brussels Calculate datetimes for shifts context = get run context date = context flow run expected start time astimezone pytz utc date if input date is None else input date date shift duration = timedelta hours=5 minutes=30 if date isoweekday == 5 else timedelta hours=8 start first shift = timezone localize datetime combine date time hour=5 minute=0 second=0 end first shift = start first shift + shift duration start first shift utc = start first shift astimezone pytz utc end first shift utc = end first shift astimezone pytz utc start second shift = end first shift end second shift = start second shift + shift duration start second shift utc = start second shift astimezone pytz utc end second shift utc = end second shift astimezone pytz utc current start = start first shift utc current end = current start + timedelta seconds=INTERVAL df = pd DataFrame from records Tags Machine machine Tags Shift int 1 Tags DowntimeIndex int 0 f Fields Speed 0 0 Timestamp start first shift utc Fields Duration 0 0 for machine in machine speed keys insert get token logger name logger pwd df while current end lt = end first shift utc data = machine data map get token user pwd machine speed keys machine speed values current start current end df = concatenate dfs df data df = process data df start first shift utc end first shift utc 1 insert get token logger name logger pwd df df Timestamp gt current start df = df df Timestamp gt = current end current start += timedelta seconds=INTERVAL current end += timedelta seconds=INTERVAL sec until next loop = current end datetime now tz=pytz utc total seconds + OFFSET if sec until next loop gt 0 logger warning Sleeping sleep sec until next loop while current end lt = end second shift utc data = machine data map get token user pwd machine speed keys machine speed values current start current end df = concatenate dfs df data df = process data df start second shift utc end second shift utc 2 insert get token logger name logger pwd df df Timestamp gt current start df = df df Timestamp gt = current end current start += timedelta seconds=INTERVAL current end += timedelta seconds=INTERVAL sec until next loop = current end datetime now tz=pytz utc total seconds + OFFSET if sec until next loop gt 0 logger warning Sleeping sleep sec until next loop ```

Mathijs Carlu · Answer

It probably could be coded more efficiently but it s a temporary solution What it does every 5 minutes 1 Requests data from our data API mapped 12 queries gt 12 tasks 2 Concatenate with previous left over data 3 Do some processing on the data 4 Send this data back to our data API The normal duration of this flow is 16 hours max It already managed to do so as well The discrepancy between times until failure is quite big as well sometimes it fails after 10 minutes sometimes only after 4 hours I originally just used local storage which worked Since I m having issues I tried to persist results to Azure Blob Storage but this hasn t really made any improvements

Mathijs Carlu · Answer

< Andreas Nigg> I also have not been able to produce a minimal working example There seems to be already an issue created for this issue though

Zanie · Answer

Hi everyone We really need a MRE to resolve this efficiently Please leave a comment in the issue if you manage to reproduce this with a simple example

Zanie · Answer

It d also be very helpful to get an idea of what versions are affected and if you are using Prefect Cloud or a local server