https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Why I’m having this error when runnin...
# ask-marvin
t
@Marvin Why I’m having this error when running a flow in a kubernetes cluster?
Copy code
2025-12-01 13:31:32,983 INFO prefect.flow_runs.runner [trace_id=0 span_id=0] ‣ Process for flow run 'conscious-pelican' exited cleanly.                                                                      
2025-12-01 13:31:33,028 WARNING prefect.FlowRunCancellingObserver [trace_id=0 span_id=0] ‣ Consumer task exited with exception                                                                               
Traceback (most recent call last):                                                                                                                                                                           
  File "/usr/local/lib/python3.12/site-packages/prefect/runner/_observers.py", line 176, in __aexit__                                                                                                        
    await self._consumer_task                                                                                                                                                                                
  File "/usr/local/lib/python3.12/site-packages/opentelemetry/instrumentation/asyncio/__init__.py", line 299, in trace_coroutine                                                                             
    return await coro                                                                                                                                                                                        
           ^^^^^^^^^^                                                                                                                                                                                        
  File "/usr/local/lib/python3.12/site-packages/prefect/runner/_observers.py", line 67, in _consume_events                                                                                                   
    async for event in self._events_subscriber:                                                                                                                                                              
  File "/usr/local/lib/python3.12/site-packages/prefect/events/clients.py", line 642, in __anext__                                                                                                           
    message = orjson.loads(await self._websocket.recv())                                                                                                                                                     
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                                                                                      
  File "/usr/local/lib/python3.12/site-packages/websockets/asyncio/connection.py", line 322, in recv                                                                                                         
    raise self.protocol.close_exc from self.recv_exc                                                                                                                                                         
websockets.exceptions.ConnectionClosedError: no close frame received or sent                                                                                                                                 
2025-12-01 13:31:33,034 ERROR asyncio [trace_id=0 span_id=0] ‣ Exception in callback FlowRunCancellingObserver._start_polling_task.<locals>.<lambda>(<Task cancell...nit__.py:276>>) at /usr/local/lib/python
handle: <Handle FlowRunCancellingObserver._start_polling_task.<locals>.<lambda>(<Task cancell...nit__.py:276>>) at /usr/local/lib/python3.12/site-packages/prefect/runner/_observers.py:99>                  
Traceback (most recent call last):                                                                                                                                                                           
  File "/usr/local/lib/python3.12/asyncio/events.py", line 88, in _run                                                                                                                                       
    self._context.run(self._callback, *self._args)                                                                                                                                                           
  File "/usr/local/lib/python3.12/site-packages/prefect/runner/_observers.py", line 103, in <lambda>                                                                                                         
    if task.exception()                                                                                                                                                                                      
       ^^^^^^^^^^^^^^^^                                                                                                                                                                                      
  File "/usr/local/lib/python3.12/site-packages/opentelemetry/instrumentation/asyncio/__init__.py", line 299, in trace_coroutine                                                                             
    return await coro                                                                                                                                                                                        
           ^^^^^^^^^^                                                                                                                                                                                        
  File "/usr/local/lib/python3.12/site-packages/prefect/utilities/services.py", line 156, in critical_service_loop                                                                                           
    await anyio.sleep(sleep)                                                                                                                                                                                 
  File "/usr/local/lib/python3.12/site-packages/anyio/_core/_eventloop.py", line 87, in sleep                                                                                                                
    return await get_async_backend().sleep(delay)                                                                                                                                                            
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^                                                                                                                                                            
  File "/usr/local/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 2369, in sleep                                                                                                            
    await sleep(delay)                                                                                                                                                                                       
  File "/usr/local/lib/python3.12/asyncio/tasks.py", line 665, in sleep                                                                                                                                      
    return await future                                                                                                                                                                                      
           ^^^^^^^^^^^^                                                                                                                                                                                      
asyncio.exceptions.CancelledError
the UI says the flow was completed successfully, though the flow code is as simple as
Copy code
@flow(name="Picanha-666")
def picanha_666() -> None:
    """
    This is a flow that will be used to test the OpenTelemetry tracing.
    """
    logger = get_run_logger()
    <http://logger.info|logger.info>("Picanha 666 flow started")
    time.sleep(10 * random.random())
    <http://logger.info|logger.info>("Picanha 666 flow ended")
m
✅ thought for 223.5 seconds
Short answer: it’s harmless. Your flow finished and the runner was shutting down; the WebSocket the runner uses to listen for “cancelling” events dropped without sending a close frame, so Prefect logged a warning during teardown. It doesn’t affect the run result. What’s happening - In Prefect 3.x, the runner uses a WebSocket subscriber (FlowRunCancellingObserver) to listen for flow-run cancellation events. - When the flow process exits, Prefect cancels that subscriber task. If the underlying WebSocket is already closed (often by an ingress/load balancer) without a proper close frame, you’ll see: - 
websockets.exceptions.ConnectionClosedError: no close frame received or sent
- followed by an 
asyncio.exceptions.CancelledError
during shutdown - Prefect falls back to polling if a WebSocket fails during operation; during shutdown it just logs the warning and continues. Your flow was already completed successfully, which matches what you see in the UI. Common causes in Kubernetes - Ingress or load balancer idle timeouts closing the WS abruptly - WebSocket upgrade not fully enabled on the ingress/proxy - Connection drained when the pod/process exits If you want to reduce or eliminate the warning - You can safely ignore it. If you want to quiet the logs: - In code:
Copy code
import logging
    logging.getLogger("prefect.FlowRunCancellingObserver").setLevel(logging.ERROR)
- Or raise overall logging level via environment variable: 
PREFECT_LOGGING_LEVEL=ERROR
- Ensure your ingress keeps WebSockets open long enough: - For NGINX Ingress, annotations like:
Copy code
<http://nginx.ingress.kubernetes.io/websocket-services|nginx.ingress.kubernetes.io/websocket-services>: prefect-api-service
    <http://nginx.ingress.kubernetes.io/proxy-read-timeout|nginx.ingress.kubernetes.io/proxy-read-timeout>: "3600"
    <http://nginx.ingress.kubernetes.io/proxy-send-timeout|nginx.ingress.kubernetes.io/proxy-send-timeout>: "3600"
    <http://nginx.ingress.kubernetes.io/proxy-connect-timeout|nginx.ingress.kubernetes.io/proxy-connect-timeout>: "600"
    <http://nginx.ingress.kubernetes.io/proxy-buffering|nginx.ingress.kubernetes.io/proxy-buffering>: "off"
    <http://nginx.ingress.kubernetes.io/proxy-request-buffering|nginx.ingress.kubernetes.io/proxy-request-buffering>: "off"
- For AWS ALB/ELB, increase the idle timeout (e.g., 90s or more). - This only impacts the event stream used for cancellation monitoring; the flow run execution and state reporting are unaffected. Related reports - WebSocket closed without close frame (worker startup / proxy): prefecthq/prefect#18749 - Long watch sessions eventually hit WS reset; run still fine: prefecthq/prefect#18428 Questions to help tailor guidance - Are you on Prefect Cloud or self-hosted Prefect Server? - Which ingress/controller are you using (nginx, ALB, Istio, etc.)? - Do you see this only at the end of runs, or during long-running flows too? Side note on your code: using 
time.sleep
is fine for a simple demo like this. If you move to async tasks/concurrency later, prefer an async sleep (e.g., 
await anyio.sleep(...)
) inside async tasks to avoid blocking the event loop.
t
@Marvin it’s a self-hosted prefect server. I’m currently running the server locally in microk8s to test the kubernetes integration
m
✅ thought for 34.3 seconds
Got it — thanks for the context. Since you’re on a self-hosted Prefect Server in microk8s, this warning is typically your ingress or connection path dropping the WebSocket abruptly when the flow finishes. The run is complete, then the cancellation listener’s WebSocket closes without a proper close frame, so Prefect logs that warning during teardown. If you’re using microk8s ingress (NGINX Ingress): - WebSockets are supported by default, but idle timeouts on the proxy often trigger this. - Bump the timeouts and turn off buffering on the API ingress to avoid early drops. Example ingress annotations you can put on the Prefect API ingress:
Copy code
<http://nginx.ingress.kubernetes.io/proxy-read-timeout|nginx.ingress.kubernetes.io/proxy-read-timeout>: "3600"
<http://nginx.ingress.kubernetes.io/proxy-send-timeout|nginx.ingress.kubernetes.io/proxy-send-timeout>: "3600"
<http://nginx.ingress.kubernetes.io/proxy-connect-timeout|nginx.ingress.kubernetes.io/proxy-connect-timeout>: "600"
<http://nginx.ingress.kubernetes.io/proxy-buffering|nginx.ingress.kubernetes.io/proxy-buffering>: "off"
<http://nginx.ingress.kubernetes.io/proxy-request-buffering|nginx.ingress.kubernetes.io/proxy-request-buffering>: "off"
Notes - You don’t need a special “websocket-services” annotation for the community NGINX Ingress; Upgrade headers are handled automatically. - If you’re using kubectl port-forward instead of ingress, these warnings are common — port-forwarded WebSockets can drop unexpectedly. Prefer an ingress or NodePort for persistent WS connections. You can safely ignore the warning - It doesn’t affect run outcomes. Prefect will fall back to polling if the WebSocket actually fails during operation; during shutdown it just logs the warning. - If you want to quiet it locally:
Copy code
import logging
  logging.getLogger("prefect.FlowRunCancellingObserver").setLevel(logging.ERROR)
or set 
PREFECT_LOGGING_LEVEL=ERROR
. Related reports for reference: - prefecthq/prefect#18749 - prefecthq/prefect#18428 A couple quick questions so I can tailor the exact steps: - Are you exposing Prefect Server via the microk8s NGINX Ingress, or using 
kubectl port-forward
? - Are you deploying with the Helm chart or your own manifests? - Do you see this only at the end of runs, or also mid-run for long-running flows?
19 Views
Open in Slack
PreviousNext
Powered by